diff --git a/asr_event_monitor_confirmed_technical_solution.md b/asr_event_monitor_confirmed_technical_solution.md index bf1dec2..d3c9fe9 100644 --- a/asr_event_monitor_confirmed_technical_solution.md +++ b/asr_event_monitor_confirmed_technical_solution.md @@ -51,6 +51,7 @@ - 保存成功后重建 AC 自动机并立即生效 - 提供单页面配置与测试 - JVM 内存保存通话状态 +- 提供通话主动关闭接口 - TTL 自动清理过期会话 - 单实例部署 @@ -64,7 +65,6 @@ - Dify 实时确认 - WebSocket 或 SSE - 通话状态查询接口 -- 通话关闭接口 - 独立 `eventId` - 多实例部署 - 复杂草稿、审核和发布流程 @@ -75,7 +75,7 @@ ### 3.1 实时检测接口 -只保留一个实时业务接口： +实时检测业务接口： ```http POST /api/v1/asr-events @@ -104,18 +104,21 @@ GET /api/v1/calls/{callId}/state - 本次新增提醒：`newAlerts` - 当前通话累计结果：`currentResults` -### 3.3 不提供通话关闭接口 +### 3.3 提供通话关闭接口 -不提供： +通话结束时，前端应主动关闭会话，立即释放内存中的通话状态： ```http POST /api/v1/calls/{callId}/close ``` +同时保留 TTL 定时清理，作为兜底机制，处理前端未调用 close、进程异常退出等遗漏场景。 + 约束条件： - 每通电话的 `callId` 必须唯一 -- 通话状态通过 TTL 自动清理 +- 正常结束路径优先调用 close +- 未关闭或异常遗留的会话由 TTL 自动清理 ### 3.4 保留每通电话独立锁 @@ -261,15 +264,18 @@ demo/ │ │ │ │ │ ├── api/ │ │ │ ├── AsrEventController.java + │ │ │ ├── CallController.java │ │ │ └── RuleAdminController.java │ │ │ │ │ ├── application/ │ │ │ ├── AsrEventMonitorService.java + │ │ │ ├── CallSessionService.java │ │ │ └── RuleManagementService.java │ │ │ │ │ ├── domain/ │ │ │ ├── AsrFinalEventRequest.java │ │ │ ├── MonitorResponse.java + │ │ │ ├── CloseCallResponse.java │ │ │ ├── MatchResult.java │ │ │ ├── AlertResult.java │ │ │ ├── RuleDocument.java @@ -758,9 +764,76 @@ try { --- -## 15. 会话 TTL 清理 +## 15. 会话关闭与 TTL 清理 -由于不提供 close 接口，会话状态依靠 TTL 清理。 +会话释放采用 **主动关闭 + TTL 兜底** 双路径： + +```text +通话结束 + → 优先：POST /api/v1/calls/{callId}/close（立即释放） + → 兜底：TTL 定时任务清理遗漏会话 +``` + +### 15.1 通话关闭接口 + +```http +POST /api/v1/calls/{callId}/close +``` + +响应示例： + +```json +{ + "callId": "call-1001", + "closed": true, + "existed": true +} +``` + +| 字段 | 说明 | +|---|---| +| `callId` | 被关闭的通话 ID | +| `closed` | 本次调用是否完成关闭处理（接口幂等，恒为 `true`） | +| `existed` | 关闭前会话是否存在；重复关闭或不存在时为 `false` | + +处理流程： + +```text +获取 callId 独立锁 +→ 删除 CallSession +→ 删除对应 ReentrantLock +→ 返回 closed=true +``` + +语义约定： + +- 接口幂等：同一 `callId` 多次 close 不报错 +- 通话结束后应尽快调用，避免会话长期占用内存 +- 由于 `callId` 必须全局唯一，关闭后不应再复用同一 `callId` 提交 ASR Final + +服务层示意： + +```java +public CloseCallResponse close(String callId) { + ReentrantLock lock = sessionStore.getLock(callId); + lock.lock(); + try { + boolean existed = sessionStore.remove(callId); + sessionStore.removeLock(callId); + return new CloseCallResponse(callId, true, existed); + } finally { + lock.unlock(); + } +} +``` + +### 15.2 TTL 兜底清理 + +即使提供 close，仍保留 TTL，用于清理以下遗留会话： + +- 前端未调用 close +- 页面异常关闭、网络故障导致 close 未送达 +- 进程重启前未及时释放的会话（单实例内存方案下重启后本就为空，TTL 主要覆盖运行期遗漏） ```yaml monitor: @@ -784,11 +857,13 @@ public void cleanupExpiredSessions() { } ``` -清理时同时删除： +TTL 清理时同样删除： - `CallSession` - 对应的 `ReentrantLock` +close 与 TTL 互不冲突：已 close 的会话不存在，TTL 跳过；未 close 的会话到期后由 TTL 删除。 + --- ## 16. 规则数据模型 @@ -1306,11 +1381,12 @@ java \ 8. `EMS`、`ems`、`E M S` 均可命中 9. 相邻 Final 拼接后可命中被切分的品牌 10. 不同通话状态相互隔离 -11. TTL 能清理过期会话 -12. 保存新规则后 Demo 接口立即使用新版本 -13. 保存失败时旧规则继续有效 -14. 两个管理页面同时编辑时能够检测版本冲突 -15. Actuator 健康检查正常 +11. close 能立即释放通话会话，重复 close 幂等 +12. TTL 能清理未 close 的过期会话 +13. 保存新规则后 Demo 接口立即使用新版本 +14. 保存失败时旧规则继续有效 +15. 两个管理页面同时编辑时能够检测版本冲突 +16. Actuator 健康检查正常 --- @@ -1414,11 +1490,19 @@ Demo 页面立即验证 ```text POST /api/v1/asr-events +POST /api/v1/calls/{callId}/close GET /api/v1/admin/rules PUT /api/v1/admin/rules POST /api/v1/admin/rules/test ``` +会话清理： + +```text +通话结束 → close 立即释放 +遗漏会话 → TTL 定时兜底清理 +``` + 最终采用： ```text diff --git a/src/main/java/com/example/demo/DemoApplication.java b/src/main/java/com/example/demo/DemoApplication.java index ebbba46..e356018 100644 --- a/src/main/java/com/example/demo/DemoApplication.java +++ b/src/main/java/com/example/demo/DemoApplication.java @@ -4,9 +4,11 @@ import com.example.demo.config.MonitorProperties; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableConfigurationProperties(MonitorProperties.class) +@EnableScheduling public class DemoApplication { public static void main(String[] args) { diff --git a/src/main/java/com/example/demo/api/CallController.java b/src/main/java/com/example/demo/api/CallController.java new file mode 100644 index 0000000..e5e83f3 --- /dev/null +++ b/src/main/java/com/example/demo/api/CallController.java @@ -0,0 +1,25 @@ +package com.example.demo.api; + +import com.example.demo.application.CallSessionService; +import com.example.demo.domain.CloseCallResponse; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequestMapping("/api/v1/calls") +public class CallController { + + private final CallSessionService callSessionService; + + public CallController(CallSessionService callSessionService) { + this.callSessionService = callSessionService; + } + + @PostMapping("/{callId}/close") + public ResponseEntity close(@PathVariable String callId) { + return ResponseEntity.ok(callSessionService.close(callId)); + } +} diff --git a/src/main/java/com/example/demo/application/CallSessionService.java b/src/main/java/com/example/demo/application/CallSessionService.java new file mode 100644 index 0000000..164446e --- /dev/null +++ b/src/main/java/com/example/demo/application/CallSessionService.java @@ -0,0 +1,31 @@ +package com.example.demo.application; + +import com.example.demo.domain.CloseCallResponse; +import com.example.demo.repository.SessionStore; +import java.util.concurrent.locks.ReentrantLock; +import org.springframework.stereotype.Service; + +@Service +public class CallSessionService { + + private final SessionStore sessionStore; + + public CallSessionService(SessionStore sessionStore) { + this.sessionStore = sessionStore; + } + + /** + * Idempotently close a call session and release its lock mapping. + */ + public CloseCallResponse close(String callId) { + ReentrantLock lock = sessionStore.getLock(callId); + lock.lock(); + try { + boolean existed = sessionStore.remove(callId); + sessionStore.removeLock(callId); + return new CloseCallResponse(callId, true, existed); + } finally { + lock.unlock(); + } + } +} diff --git a/src/main/java/com/example/demo/domain/CloseCallResponse.java b/src/main/java/com/example/demo/domain/CloseCallResponse.java new file mode 100644 index 0000000..af1f58e --- /dev/null +++ b/src/main/java/com/example/demo/domain/CloseCallResponse.java @@ -0,0 +1,8 @@ +package com.example.demo.domain; + +public record CloseCallResponse( + String callId, + boolean closed, + boolean existed +) { +} diff --git a/src/main/java/com/example/demo/job/SessionCleanupJob.java b/src/main/java/com/example/demo/job/SessionCleanupJob.java new file mode 100644 index 0000000..f52b676 --- /dev/null +++ b/src/main/java/com/example/demo/job/SessionCleanupJob.java @@ -0,0 +1,39 @@ +package com.example.demo.job; + +import com.example.demo.config.MonitorProperties; +import com.example.demo.repository.SessionStore; +import java.time.Clock; +import java.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Component +public class SessionCleanupJob { + + private static final Logger log = LoggerFactory.getLogger(SessionCleanupJob.class); + + private final SessionStore sessionStore; + private final MonitorProperties properties; + private final Clock clock; + + public SessionCleanupJob( + SessionStore sessionStore, + MonitorProperties properties, + Clock clock + ) { + this.sessionStore = sessionStore; + this.properties = properties; + this.clock = clock; + } + + @Scheduled(fixedDelayString = "${monitor.session-cleanup-interval:PT10M}") + public void cleanupExpiredSessions() { + Instant threshold = Instant.now(clock).minus(properties.sessionTtl()); + int removed = sessionStore.removeExpired(threshold); + if (removed > 0) { + log.info("Removed {} expired call sessions older than {}", removed, threshold); + } + } +} diff --git a/src/main/java/com/example/demo/repository/InMemorySessionStore.java b/src/main/java/com/example/demo/repository/InMemorySessionStore.java new file mode 100644 index 0000000..e53c36d --- /dev/null +++ b/src/main/java/com/example/demo/repository/InMemorySessionStore.java @@ -0,0 +1,88 @@ +package com.example.demo.repository; + +import com.example.demo.domain.CallSession; +import java.time.Clock; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; +import org.springframework.stereotype.Component; + +@Component +public class InMemorySessionStore implements SessionStore { + + private final ConcurrentHashMap sessions = new ConcurrentHashMap<>(); + private final ConcurrentHashMap callLocks = new ConcurrentHashMap<>(); + private final Clock clock; + + public InMemorySessionStore(Clock clock) { + this.clock = clock; + } + + @Override + public ReentrantLock getLock(String callId) { + return callLocks.computeIfAbsent(callId, ignored -> new ReentrantLock()); + } + + @Override + public CallSession getOrCreate(String callId) { + return sessions.computeIfAbsent(callId, id -> new CallSession(id, clock)); + } + + @Override + public Optional find(String callId) { + return Optional.ofNullable(sessions.get(callId)); + } + + @Override + public boolean remove(String callId) { + return sessions.remove(callId) != null; + } + + @Override + public void removeLock(String callId) { + callLocks.compute(callId, (id, existing) -> { + if (existing == null) { + return null; + } + // Drop only when no session remains and no waiters remain after current holder unlocks. + if (!sessions.containsKey(id) + && !existing.hasQueuedThreads() + && (existing.isHeldByCurrentThread() || !existing.isLocked())) { + return null; + } + return existing; + }); + } + + @Override + public int removeExpired(Instant threshold) { + int removed = 0; + List callIds = new ArrayList<>(sessions.keySet()); + + for (String callId : callIds) { + ReentrantLock lock = getLock(callId); + lock.lock(); + try { + CallSession session = sessions.get(callId); + if (session != null && session.lastAccessed().isBefore(threshold)) { + if (sessions.remove(callId, session)) { + removed++; + } + } + } finally { + lock.unlock(); + } + removeLock(callId); + } + + return removed; + } + + @Override + public int size() { + return sessions.size(); + } +} diff --git a/src/main/java/com/example/demo/repository/SessionStore.java b/src/main/java/com/example/demo/repository/SessionStore.java new file mode 100644 index 0000000..a91e17b --- /dev/null +++ b/src/main/java/com/example/demo/repository/SessionStore.java @@ -0,0 +1,38 @@ +package com.example.demo.repository; + +import com.example.demo.domain.CallSession; +import java.time.Instant; +import java.util.Optional; +import java.util.concurrent.locks.ReentrantLock; + +public interface SessionStore { + + ReentrantLock getLock(String callId); + + CallSession getOrCreate(String callId); + + Optional find(String callId); + + /** + * Remove the session for callId if present. + * + * @return true if a session existed and was removed + */ + boolean remove(String callId); + + /** + * Remove the per-call lock mapping when idle and no session remains. + * Safe to call while holding that same lock instance. + */ + void removeLock(String callId); + + /** + * Remove sessions whose lastAccessed is strictly before threshold, + * along with their per-call locks. + * + * @return number of sessions removed + */ + int removeExpired(Instant threshold); + + int size(); +} diff --git a/src/test/java/com/example/demo/api/CallControllerTest.java b/src/test/java/com/example/demo/api/CallControllerTest.java new file mode 100644 index 0000000..985f2b0 --- /dev/null +++ b/src/test/java/com/example/demo/api/CallControllerTest.java @@ -0,0 +1,40 @@ +package com.example.demo.api; + +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import com.example.demo.repository.SessionStore; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import org.springframework.web.context.WebApplicationContext; + +@SpringBootTest +class CallControllerTest { + + @Autowired + private WebApplicationContext context; + + @Autowired + private SessionStore sessionStore; + + @Test + void closeEndpointIsIdempotent() throws Exception { + MockMvc mockMvc = MockMvcBuilders.webAppContextSetup(context).build(); + sessionStore.getOrCreate("call-1001"); + + mockMvc.perform(post("/api/v1/calls/call-1001/close")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.callId").value("call-1001")) + .andExpect(jsonPath("$.closed").value(true)) + .andExpect(jsonPath("$.existed").value(true)); + + mockMvc.perform(post("/api/v1/calls/call-1001/close")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.closed").value(true)) + .andExpect(jsonPath("$.existed").value(false)); + } +} diff --git a/src/test/java/com/example/demo/application/CallSessionServiceTest.java b/src/test/java/com/example/demo/application/CallSessionServiceTest.java new file mode 100644 index 0000000..4a43b0f --- /dev/null +++ b/src/test/java/com/example/demo/application/CallSessionServiceTest.java @@ -0,0 +1,47 @@ +package com.example.demo.application; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.example.demo.domain.CloseCallResponse; +import com.example.demo.repository.InMemorySessionStore; +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class CallSessionServiceTest { + + private InMemorySessionStore store; + private CallSessionService service; + + @BeforeEach + void setUp() { + store = new InMemorySessionStore(Clock.fixed(Instant.parse("2026-07-15T00:00:00Z"), ZoneOffset.UTC)); + service = new CallSessionService(store); + } + + @Test + void closeReleasesExistingSessionIdempotently() { + store.getOrCreate("call-1001"); + + CloseCallResponse first = service.close("call-1001"); + assertEquals("call-1001", first.callId()); + assertTrue(first.closed()); + assertTrue(first.existed()); + assertEquals(0, store.size()); + + CloseCallResponse second = service.close("call-1001"); + assertTrue(second.closed()); + assertFalse(second.existed()); + } + + @Test + void closeUnknownCallStillSucceeds() { + CloseCallResponse response = service.close("missing"); + assertTrue(response.closed()); + assertFalse(response.existed()); + } +} diff --git a/src/test/java/com/example/demo/job/SessionCleanupJobTest.java b/src/test/java/com/example/demo/job/SessionCleanupJobTest.java new file mode 100644 index 0000000..147315e --- /dev/null +++ b/src/test/java/com/example/demo/job/SessionCleanupJobTest.java @@ -0,0 +1,39 @@ +package com.example.demo.job; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.example.demo.config.MonitorProperties; +import com.example.demo.repository.SessionStore; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class SessionCleanupJobTest { + + @Test + void cleanupUsesTtlThreshold() { + SessionStore sessionStore = mock(SessionStore.class); + MonitorProperties properties = new MonitorProperties( + 2, + 500, + Duration.ofHours(2), + Duration.ofMinutes(10) + ); + Instant fixed = Instant.parse("2026-07-15T12:00:00Z"); + Clock clock = Clock.fixed(fixed, ZoneOffset.UTC); + when(sessionStore.removeExpired(fixed.minus(Duration.ofHours(2)))).thenReturn(3); + + SessionCleanupJob job = new SessionCleanupJob(sessionStore, properties, clock); + job.cleanupExpiredSessions(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(Instant.class); + verify(sessionStore).removeExpired(captor.capture()); + assertEquals(fixed.minus(Duration.ofHours(2)), captor.getValue()); + } +} diff --git a/src/test/java/com/example/demo/repository/InMemorySessionStoreTest.java b/src/test/java/com/example/demo/repository/InMemorySessionStoreTest.java new file mode 100644 index 0000000..df83ccc --- /dev/null +++ b/src/test/java/com/example/demo/repository/InMemorySessionStoreTest.java @@ -0,0 +1,113 @@ +package com.example.demo.repository; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.example.demo.domain.CallSession; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantLock; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class InMemorySessionStoreTest { + + private AtomicReference now; + private InMemorySessionStore store; + + @BeforeEach + void setUp() { + now = new AtomicReference<>(Instant.parse("2026-07-15T00:00:00Z")); + Clock clock = new Clock() { + @Override + public ZoneOffset getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(java.time.ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now.get(); + } + }; + store = new InMemorySessionStore(clock); + } + + @Test + void getOrCreateReturnsSameSessionPerCallId() { + CallSession a = store.getOrCreate("call-1"); + CallSession b = store.getOrCreate("call-1"); + CallSession c = store.getOrCreate("call-2"); + + assertSame(a, b); + assertNotSame(a, c); + assertEquals(2, store.size()); + } + + @Test + void getLockReturnsSameLockPerCallId() { + ReentrantLock a = store.getLock("call-1"); + ReentrantLock b = store.getLock("call-1"); + assertSame(a, b); + assertNotSame(a, store.getLock("call-2")); + } + + @Test + void removeExpiredDeletesSessionAndAllowsNewLock() { + CallSession session = store.getOrCreate("call-1"); + ReentrantLock oldLock = store.getLock("call-1"); + session.touch(Clock.fixed(now.get(), ZoneOffset.UTC)); + + now.set(now.get().plus(Duration.ofHours(3))); + Instant threshold = now.get().minus(Duration.ofHours(2)); + + assertEquals(1, store.removeExpired(threshold)); + assertEquals(0, store.size()); + assertTrue(store.find("call-1").isEmpty()); + + ReentrantLock newLock = store.getLock("call-1"); + assertNotSame(oldLock, newLock); + } + + @Test + void removeDeletesSessionImmediately() { + store.getOrCreate("call-1"); + ReentrantLock oldLock = store.getLock("call-1"); + + assertTrue(store.remove("call-1")); + store.removeLock("call-1"); + + assertEquals(0, store.size()); + assertFalse(store.remove("call-1")); + assertNotSame(oldLock, store.getLock("call-1")); + } + + @Test + void removeExpiredKeepsFreshSessions() { + store.getOrCreate("fresh"); + now.set(now.get().plus(Duration.ofMinutes(30))); + Instant threshold = now.get().minus(Duration.ofHours(2)); + + assertEquals(0, store.removeExpired(threshold)); + assertEquals(1, store.size()); + } + + @Test + void differentCallsAreIsolated() { + store.getOrCreate("call-a").appendCitizenFinal(1, "顺丰", 2); + store.getOrCreate("call-b").appendCitizenFinal(1, "拼多多", 2); + + assertEquals("顺丰", store.find("call-a").orElseThrow().buildMatchText()); + assertEquals("拼多多", store.find("call-b").orElseThrow().buildMatchText()); + } +}