mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 11:13:43 +08:00
fix(webchat): detach SSE subscribers and reclaim orphan runs
Use detach rather than complete for WebChat SSE disconnect callbacks, add an orphan-run grace policy for subscriberless active runs, and emergency-save partial assistant output when reclaimed.
This commit is contained in:
倪程伟
GitHub
parent
e9c08edb3a
commit
55cf53b9b4
@ -226,6 +226,18 @@ public class ChatStreamTracker {
|
||||
*/
|
||||
volatile long lastEventAt = System.currentTimeMillis();
|
||||
|
||||
/**
|
||||
* Wall-clock millis at which the subscriber list last became empty
|
||||
* while the run was still alive (not done). Null when there is at
|
||||
* least one subscriber, or when the run already finished via the
|
||||
* normal {@code done} path. Drives the orphan-grace eviction in
|
||||
* {@link ChatStreamTracker#cleanupStaleRuns()} (issue #587): a run
|
||||
* whose only subscriber disconnected is invisible to its owner and
|
||||
* unreachable (webchat has no re-attach endpoint), so it is torn down
|
||||
* after a grace period instead of burning tokens until the idle sweep.
|
||||
*/
|
||||
volatile Long subscribersZeroSince;
|
||||
|
||||
/** Bound agent identifier; null while not yet resolved. */
|
||||
volatile Long agentId;
|
||||
|
||||
@ -973,6 +985,10 @@ public class ChatStreamTracker {
|
||||
// Without this, async_task_completed fired after `done` would be silently
|
||||
// dropped, leaving the chat UI stuck on the "正在生成中" placeholder.
|
||||
state.subscribers.add(emitter);
|
||||
// A (re-)attached subscriber clears the orphan clock — the run is
|
||||
// visible to its owner again, so the grace-period eviction in
|
||||
// cleanupStaleRuns() should not fire (issue #587).
|
||||
state.subscribersZeroSince = null;
|
||||
|
||||
// Deliver MCP progress snapshots on reconnect (progress events skip buffer replay)
|
||||
sendProgressSnapshots(conversationId, emitter);
|
||||
@ -1096,6 +1112,14 @@ public class ChatStreamTracker {
|
||||
}
|
||||
synchronized (state.lock) {
|
||||
state.subscribers.remove(emitter);
|
||||
// When the last subscriber leaves and the run is still alive, arm
|
||||
// the orphan clock — see RunState.subscribersZeroSince. The run
|
||||
// is now invisible to its owner and (for webchat) unreachable, so
|
||||
// cleanupStaleRuns will reclaim it after the grace window unless a
|
||||
// fresh subscriber re-attaches (which clears the clock in attach()).
|
||||
if (state.subscribers.isEmpty() && !state.done && state.subscribersZeroSince == null) {
|
||||
state.subscribersZeroSince = System.currentTimeMillis();
|
||||
}
|
||||
}
|
||||
log.debug("Emitter detached from stream: {} (remaining={})",
|
||||
conversationId, state.subscribers.size());
|
||||
@ -1538,6 +1562,24 @@ public class ChatStreamTracker {
|
||||
@org.springframework.beans.factory.annotation.Value("${mateclaw.sse.idle-timeout-minutes:30}")
|
||||
private int idleTimeoutMinutes = 30;
|
||||
|
||||
/**
|
||||
* Grace period (seconds) before an orphaned run is reclaimed. A run is
|
||||
* "orphaned" when its subscriber list has been empty since some instant
|
||||
* (the only SSE client disconnected) while the agent Flux is still
|
||||
* running — invisible to its owner and, for the WebChat channel,
|
||||
* unreachable (no re-attach endpoint). The default 2 minutes tolerates a
|
||||
* network blip + a client-side regenerate retry; once it elapses with no
|
||||
* subscriber returning, the run is disposed and its partial assistant
|
||||
* content is flushed via {@code emergencySaveCallback} (issue #587).
|
||||
* <p>
|
||||
* Note: a run that keeps producing events but has no subscribers is NOT
|
||||
* considered stuck — {@code lastEventAt} keeps it out of the idle bucket.
|
||||
* The orphan bucket specifically catches "alive but nobody's watching",
|
||||
* which the idle watchdog cannot see.
|
||||
*/
|
||||
@org.springframework.beans.factory.annotation.Value("${mateclaw.webchat.orphan-grace-sec:120}")
|
||||
private int orphanGraceSeconds = 120;
|
||||
|
||||
/**
|
||||
* Test hook — backdates the {@code lastEventAt} timestamp on an
|
||||
* existing RunState so {@link #cleanupStaleRuns()} can be exercised
|
||||
@ -1567,16 +1609,32 @@ public class ChatStreamTracker {
|
||||
this.idleTimeoutMinutes = minutes;
|
||||
}
|
||||
|
||||
/** Test hook — backdate the orphan clock on an existing run. */
|
||||
void backdateOrphanForTesting(String conversationId, long subscribersZeroSince) {
|
||||
RunState state = runs.get(conversationId);
|
||||
if (state != null) {
|
||||
state.subscribersZeroSince = subscribersZeroSince;
|
||||
}
|
||||
}
|
||||
|
||||
/** Test hook — override the orphan grace in pure-unit tests that bypass Spring. */
|
||||
void setOrphanGraceSecondsForTesting(int seconds) {
|
||||
this.orphanGraceSeconds = seconds;
|
||||
}
|
||||
|
||||
/**
|
||||
* 定期清理过期的 RunState，防止内存泄漏。
|
||||
* - 已完成超过 {@link #DONE_RETENTION_MS} 的 → 移除
|
||||
* - 自 {@link RunState#lastEventAt} 算起静默超过
|
||||
* {@link #idleTimeoutMinutes} 分钟的 → 强制移除（视为卡死）
|
||||
* - 订阅者清零超过 {@link #orphanGraceSeconds} 且仍在运行的孤儿 →
|
||||
* 移除（webchat 无重连端点，运行对调用方不可见不可达，见 #587）
|
||||
*/
|
||||
@org.springframework.scheduling.annotation.Scheduled(fixedRate = 600_000)
|
||||
public void cleanupStaleRuns() {
|
||||
long now = System.currentTimeMillis();
|
||||
long idleThresholdMs = (long) idleTimeoutMinutes * 60_000L;
|
||||
long orphanGraceMs = (long) orphanGraceSeconds * 1000L;
|
||||
int evicted = 0;
|
||||
|
||||
var iterator = runs.entrySet().iterator();
|
||||
@ -1585,6 +1643,16 @@ public class ChatStreamTracker {
|
||||
RunState state = entry.getValue();
|
||||
long age = now - state.createdAt;
|
||||
long idleMs = now - state.lastEventAt;
|
||||
Long orphanSince = state.subscribersZeroSince;
|
||||
long orphanMs = orphanSince != null ? now - orphanSince : -1L;
|
||||
// Subscriber count under the lock so the orphan decision is
|
||||
// consistent with the subscriber list (subscribersZeroSince is
|
||||
// normally null whenever a subscriber is present, but a concurrent
|
||||
// attach/detach could race the read — guard against that here).
|
||||
int subCount;
|
||||
synchronized (state.lock) {
|
||||
subCount = state.subscribers.size();
|
||||
}
|
||||
|
||||
boolean shouldEvict = false;
|
||||
String reason = null;
|
||||
@ -1592,6 +1660,16 @@ public class ChatStreamTracker {
|
||||
if (state.done && age > DONE_RETENTION_MS) {
|
||||
shouldEvict = true;
|
||||
reason = "completed and expired";
|
||||
} else if (!state.done && subCount == 0 && orphanSince != null && orphanMs > orphanGraceMs) {
|
||||
// Orphan: subscriber list empty longer than the grace window
|
||||
// while the agent Flux is still running. Invisible + (for
|
||||
// webchat) unreachable, so reclaim it instead of letting it
|
||||
// burn tokens until the idle sweep (issue #587). A run that's
|
||||
// actively producing events is NOT exempt — the whole point is
|
||||
// nobody is watching those events.
|
||||
shouldEvict = true;
|
||||
reason = "orphaned: no subscribers for " + (orphanMs / 1000)
|
||||
+ "s (grace " + orphanGraceSeconds + "s); run still active";
|
||||
} else if (idleMs > idleThresholdMs) {
|
||||
shouldEvict = true;
|
||||
reason = "idle for " + (idleMs / 1000) + "s (threshold "
|
||||
|
||||
@ -186,15 +186,25 @@ public class WebChatController {
|
||||
|
||||
log.info("[WebChat] Stream: agentId={}, conversationId={}, visitor={}", agentId, conversationId, visitorId);
|
||||
|
||||
// 注册 emitter 回调
|
||||
emitter.onCompletion(() -> log.debug("[WebChat] SSE completed: {}", conversationId));
|
||||
// Register emitter callbacks. An SSE disconnect means this subscriber
|
||||
// left, not that the agent run finished. Use detach() instead of
|
||||
// complete(); complete() would prematurely mark the RunState done,
|
||||
// drop later content deltas from the replay buffer, and double-count
|
||||
// completion when the agent Flux actually finishes.
|
||||
emitter.onCompletion(() -> {
|
||||
log.debug("[WebChat] SSE completed: {}", conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
});
|
||||
emitter.onTimeout(() -> {
|
||||
log.debug("[WebChat] SSE timeout: {}", conversationId);
|
||||
streamTracker.complete(conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
// Explicitly complete after timeout so the servlet container does
|
||||
// not rethrow AsyncRequestTimeoutException.
|
||||
emitter.complete();
|
||||
});
|
||||
emitter.onError(e -> {
|
||||
log.debug("[WebChat] SSE error: {} - {}", conversationId, e.getMessage());
|
||||
streamTracker.complete(conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
});
|
||||
|
||||
sseExecutor.execute(() -> {
|
||||
@ -320,6 +330,11 @@ public class WebChatController {
|
||||
// HTTP call keeps running — token burn + side-effect tools still fire.
|
||||
// Mirrors ChatController#chatStream line 495.
|
||||
streamTracker.setDisposable(conversationId, disposable);
|
||||
// Wire the emergency save so an orphaned run (only subscriber
|
||||
// gone) is flushed as an "interrupted" assistant message when
|
||||
// the grace-period eviction reclaims it — otherwise the visitor
|
||||
// would see only their own user message (issue #587).
|
||||
registerEmergencySave(conversationId, assistantReply, usage, modelInfo);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("[WebChat] Error: {}", e.getMessage(), e);
|
||||
@ -1242,14 +1257,18 @@ public class WebChatController {
|
||||
return emitter;
|
||||
}
|
||||
|
||||
emitter.onCompletion(() -> log.debug("[WebChat] approve SSE completed: {}", conversationId));
|
||||
emitter.onCompletion(() -> {
|
||||
log.debug("[WebChat] approve SSE completed: {}", conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
});
|
||||
emitter.onTimeout(() -> {
|
||||
log.debug("[WebChat] approve SSE timeout: {}", conversationId);
|
||||
streamTracker.complete(conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
emitter.complete();
|
||||
});
|
||||
emitter.onError(e -> {
|
||||
log.debug("[WebChat] approve SSE error: {} - {}", conversationId, e.getMessage());
|
||||
streamTracker.complete(conversationId);
|
||||
streamTracker.detach(conversationId, emitter);
|
||||
});
|
||||
|
||||
String actor = webchatUsername(visitorId);
|
||||
@ -1372,6 +1391,7 @@ public class WebChatController {
|
||||
})
|
||||
.subscribe();
|
||||
streamTracker.setDisposable(conversationId, disposable);
|
||||
registerEmergencySave(conversationId, assistantReply, usage, modelInfo);
|
||||
} catch (Exception e) {
|
||||
log.error("[WebChat] approve failed for {}: {}", conversationId, e.getMessage());
|
||||
try {
|
||||
@ -1638,6 +1658,53 @@ public class WebChatController {
|
||||
return parts;
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an emergency-save callback that flushes the partial assistant
|
||||
* reply accumulated so far as an {@code interrupted} message. Wired on
|
||||
* both {@code /stream} and {@code /sessions/approve} so that when a run is
|
||||
* reclaimed while its only subscriber is gone (orphan-grace eviction,
|
||||
* shutdown, admin force-recycle), the visitor can still retrieve the
|
||||
* partial answer via {@code /sessions/messages} instead of seeing only the
|
||||
* user message (issue #587). Mirrors ChatController#emergencySaveAccumulator.
|
||||
*
|
||||
* @param conversationId target conversation
|
||||
* @param assistantReply live accumulator appended to in doOnNext
|
||||
* @param usage [prompt, completion, cacheRead, cacheWrite, reasoning]
|
||||
* @param modelInfo [runtimeModel, runtimeProvider]
|
||||
*/
|
||||
private void registerEmergencySave(String conversationId, StringBuilder assistantReply,
|
||||
int[] usage, String[] modelInfo) {
|
||||
streamTracker.setEmergencySaveCallback(conversationId, () -> {
|
||||
try {
|
||||
String reply = assistantReply.toString();
|
||||
if (reply.isBlank()) {
|
||||
log.debug("[WebChat] Emergency save skipped (empty reply): {}", conversationId);
|
||||
return;
|
||||
}
|
||||
// usage length varies by call site (chatStream = 5 tokens,
|
||||
// approve replay = 2); read defensively so the save never
|
||||
// throws ArrayIndexOutOfBoundsException.
|
||||
int prompt = usage.length > 0 ? usage[0] : 0;
|
||||
int completion = usage.length > 1 ? usage[1] : 0;
|
||||
int cacheRead = usage.length > 2 ? usage[2] : 0;
|
||||
int cacheWrite = usage.length > 3 ? usage[3] : 0;
|
||||
int reasoning = usage.length > 4 ? usage[4] : 0;
|
||||
String runtimeModel = modelInfo.length > 0 ? modelInfo[0] : null;
|
||||
String runtimeProvider = modelInfo.length > 1 ? modelInfo[1] : null;
|
||||
conversationService.saveMessage(
|
||||
conversationId, "assistant", reply, List.of(),
|
||||
"interrupted",
|
||||
prompt, completion, cacheRead, cacheWrite, reasoning,
|
||||
runtimeModel, runtimeProvider, null);
|
||||
log.info("[WebChat] Emergency-saved partial assistant reply: " +
|
||||
"conversationId={}, textLen={}",
|
||||
conversationId, reply.length());
|
||||
} catch (Exception e) {
|
||||
log.warn("[WebChat] Emergency save failed for {}: {}", conversationId, e.getMessage());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// ==================== 内部方法 ====================
|
||||
|
||||
private static final Pattern SESSION_ID_PATTERN = Pattern.compile("[A-Za-z0-9_-]{1,64}");
|
||||
|
||||
@ -0,0 +1,111 @@
|
||||
package vip.mate.channel.web;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Pins the {@link ChatStreamTracker#detach(String, SseEmitter)} contract: a
|
||||
* subscriber going away (SSE timeout / error / client close) must NOT mark the
|
||||
* run as done. This is the tracker-level root of WebChatController issue #587
|
||||
* defect 1 — the controller previously called {@code complete()} from its
|
||||
* {@code onTimeout}/{@code onError} callbacks, which polluted RunState ahead
|
||||
* of the agent finishing and dropped subsequent content deltas from the replay
|
||||
* buffer.
|
||||
*
|
||||
* <p>These tests assert the tracker-side invariant the controller now relies on:
|
||||
* detach only removes the subscriber, the run keeps running, and events still
|
||||
* reach the buffer for any re-attaching subscriber.
|
||||
*/
|
||||
class ChatStreamTrackerDetachSemanticsTest {
|
||||
|
||||
private ChatStreamTracker newTracker() {
|
||||
return new ChatStreamTracker(new ObjectMapper());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("detach() leaves the run running — isRunning() stays true")
|
||||
void detachKeepsRunRunning() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "detach-running";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
SseEmitter emitter = new SseEmitter();
|
||||
tracker.attach(cid, emitter);
|
||||
// Simulate the SSE onTimeout path: this is what WebChatController now calls.
|
||||
tracker.detach(cid, emitter);
|
||||
|
||||
assertTrue(tracker.isRunning(cid),
|
||||
"detach only removes the subscriber; the run must stay running");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("detach() does NOT mark done — subsequent events still buffer for replay")
|
||||
void detachDoesNotMarkDone() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "detach-buffer";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
SseEmitter gone = new SseEmitter();
|
||||
tracker.attach(cid, gone);
|
||||
tracker.detach(cid, gone);
|
||||
|
||||
// After the subscriber left, the agent keeps producing. These events must
|
||||
// land in the buffer so a re-attaching subscriber can replay them — the
|
||||
// whole point of not prematurely calling complete().
|
||||
tracker.broadcast(cid, "content_delta", "{\"text\":\"still-alive\"}");
|
||||
|
||||
// A fresh subscriber attaching should be able to see the buffered event
|
||||
// (proving done was NOT set, which would have dropped the broadcast).
|
||||
SseEmitter late = new SseEmitter();
|
||||
AtomicInteger received = new AtomicInteger();
|
||||
late.onCompletion(() -> {
|
||||
});
|
||||
// attach replays the buffer synchronously; we can't easily count sends on a
|
||||
// raw SseEmitter, but the key assertion is that attach returns true (state
|
||||
// exists and is not in a terminal window that drops events).
|
||||
assertTrue(tracker.attach(cid, late), "attach must succeed — run is still alive");
|
||||
assertTrue(tracker.isRunning(cid));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("contrast: complete() DOES mark done (the old, buggy behavior)")
|
||||
void completeMarksDone() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "complete-done";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
// complete() is the agent-finished path — it should mark the run done.
|
||||
tracker.complete(cid);
|
||||
assertFalse(tracker.isRunning(cid),
|
||||
"complete() is the real finish signal; detach() must NOT be");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("detach is idempotent and safe when no run exists")
|
||||
void detachSafeWhenAbsent() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
SseEmitter emitter = new SseEmitter();
|
||||
// No run registered — detach must not throw.
|
||||
tracker.detach("never-registered", emitter);
|
||||
|
||||
tracker.register("present");
|
||||
tracker.attach("present", emitter);
|
||||
// Detaching twice must be a no-op the second time.
|
||||
tracker.detach("present", emitter);
|
||||
tracker.detach("present", emitter);
|
||||
// run still alive
|
||||
assertTrue(tracker.isRunning("present"));
|
||||
assertEquals(0, tracker.getAllSnapshot().getFirst().subscriberCount());
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,149 @@
|
||||
package vip.mate.channel.web;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Pins the orphan-run policy added for issue #587: when a run's only
|
||||
* subscriber disconnects (webchat SSE timeout/error) while the agent Flux is
|
||||
* still running, the run becomes invisible + unreachable and must be reclaimed
|
||||
* after a grace window instead of burning tokens until the 30-min idle sweep.
|
||||
*
|
||||
* <p>Composes with the detach() fix from #587 defect 1: detach() arms the
|
||||
* orphan clock when the subscriber list empties; attach()/closeSubscribers()
|
||||
* clear it.
|
||||
*/
|
||||
class ChatStreamTrackerOrphanPolicyTest {
|
||||
|
||||
private ChatStreamTracker newTracker() {
|
||||
ChatStreamTracker t = new ChatStreamTracker(new ObjectMapper());
|
||||
t.setIdleTimeoutMinutesForTesting(30); // keep the idle bucket out of the way
|
||||
t.setOrphanGraceSecondsForTesting(2); // tight grace for unit-test speed
|
||||
return t;
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Orphan run (no subscribers past grace) is evicted")
|
||||
void orphanRunEvictedAfterGrace() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "orphan-evict";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
// No subscriber ever attached — simulate the clock by backdating, which
|
||||
// is exactly what detach() would have set when the last subscriber left.
|
||||
tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); // 3s > 2s grace
|
||||
|
||||
tracker.cleanupStaleRuns();
|
||||
|
||||
assertFalse(tracker.hasRunStateForTesting(cid),
|
||||
"orphan run past the grace window must be evicted");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Orphan run within grace survives")
|
||||
void orphanRunWithinGraceSurvives() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "orphan-fresh";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
// Just became orphaned — 1s, within the 2s grace.
|
||||
tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 1_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
|
||||
assertTrue(tracker.hasRunStateForTesting(cid),
|
||||
"orphan run inside the grace window must survive — tolerates a brief disconnect");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("Orphan eviction fires emergencySaveCallback before dispose")
|
||||
void orphanEvictionFiresEmergencySave() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "orphan-save";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
AtomicInteger saveCount = new AtomicInteger();
|
||||
tracker.setEmergencySaveCallback(cid, saveCount::incrementAndGet);
|
||||
|
||||
tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
|
||||
assertEquals(1, saveCount.get(),
|
||||
"partial assistant content must be flushed via the emergency " +
|
||||
"save before the orphan run is disposed");
|
||||
assertFalse(tracker.hasRunStateForTesting(cid));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("A done run is never treated as an orphan")
|
||||
void doneRunNotOrphan() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "done-not-orphan";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
tracker.complete(cid); // mark done
|
||||
|
||||
// Even with an orphan clock backdated past grace, a done run must not
|
||||
// hit the orphan branch (it's finalized via the done path / retention).
|
||||
tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
|
||||
// done run is kept for DONE_RETENTION_MS (5 min) — still here right after.
|
||||
assertTrue(tracker.hasRunStateForTesting(cid),
|
||||
"a done run must not be evicted as an orphan — it's already finalized");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("A run with a live subscriber is never an orphan")
|
||||
void runWithSubscriberNotOrphan() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "has-sub";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
SseEmitter em = new SseEmitter();
|
||||
tracker.attach(cid, em); // attaches a subscriber -> clears orphan clock
|
||||
|
||||
// Backdate would-be orphan clock — but a subscriber is present, so the
|
||||
// orphan branch must not fire regardless.
|
||||
tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L);
|
||||
tracker.cleanupStaleRuns();
|
||||
|
||||
assertTrue(tracker.hasRunStateForTesting(cid),
|
||||
"a run with a live subscriber must never be evicted as an orphan");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("detach() arms the orphan clock; re-attach clears it")
|
||||
void detachArmsClockAttachClears() {
|
||||
ChatStreamTracker tracker = newTracker();
|
||||
String cid = "detach-rearm";
|
||||
tracker.register(cid);
|
||||
tracker.incrementFlux(cid);
|
||||
|
||||
SseEmitter em = new SseEmitter();
|
||||
tracker.attach(cid, em);
|
||||
// Detach the only subscriber -> clock armed, run still running.
|
||||
tracker.detach(cid, em);
|
||||
assertTrue(tracker.isRunning(cid),
|
||||
"run must still be running after the only subscriber detaches");
|
||||
|
||||
// A fresh subscriber re-attaches within grace -> clock cleared, survives.
|
||||
SseEmitter reattached = new SseEmitter();
|
||||
assertTrue(tracker.attach(cid, reattached));
|
||||
tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); // would-be past grace
|
||||
tracker.cleanupStaleRuns();
|
||||
assertTrue(tracker.hasRunStateForTesting(cid),
|
||||
"re-attach clears the orphan clock even if it was backdated");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user