diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java index 6e6e606..f58cc41 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java @@ -30,6 +30,8 @@ import com.imeeting.service.biz.MeetingPointsService; import com.imeeting.service.biz.MeetingSummaryFileService; import com.imeeting.service.biz.MeetingTranscriptChapterService; import com.imeeting.service.biz.MeetingTranscriptFileService; +import com.imeeting.support.retry.RetryExecutor; +import com.imeeting.support.retry.RetryOptions; import com.imeeting.support.TaskSecurityContextRunner; import com.imeeting.support.redis.MeetingAsrPermitCache; @@ -105,6 +107,7 @@ public class AiTaskServiceImpl extends ServiceImpl impleme private final TaskSecurityContextRunner taskSecurityContextRunner; private final MeetingExternalSummaryWebhookTrigger meetingExternalSummaryWebhookTrigger; private final SysParamService sysParamService; + private final RetryExecutor retryExecutor; @Autowired @Qualifier("asrTaskExecutor") @@ -152,6 +155,31 @@ public class AiTaskServiceImpl extends ServiceImpl impleme TaskSecurityContextRunner taskSecurityContextRunner, MeetingExternalSummaryWebhookTrigger meetingExternalSummaryWebhookTrigger, SysParamService sysParamService) { + this(meetingMapper, transcriptMapper, aiModelService, objectMapper, sysUserMapper, hotWordService, + meetingLockCache, meetingAsrPermitCache, meetingProgressService, meetingPointsService, + meetingSummaryFileService, meetingTranscriptFileService, meetingTranscriptChapterService, + meetingSummaryPromptAssembler, taskSecurityContextRunner, meetingExternalSummaryWebhookTrigger, + sysParamService, new RetryExecutor()); + } + + public AiTaskServiceImpl(MeetingMapper meetingMapper, + MeetingTranscriptMapper transcriptMapper, + AiModelService aiModelService, + ObjectMapper objectMapper, + SysUserMapper sysUserMapper, + HotWordService hotWordService, + MeetingLockCache meetingLockCache, + MeetingAsrPermitCache meetingAsrPermitCache, + MeetingProgressService meetingProgressService, + MeetingPointsService meetingPointsService, + MeetingSummaryFileService meetingSummaryFileService, + MeetingTranscriptFileService meetingTranscriptFileService, + MeetingTranscriptChapterService meetingTranscriptChapterService, + MeetingSummaryPromptAssembler meetingSummaryPromptAssembler, + TaskSecurityContextRunner taskSecurityContextRunner, + MeetingExternalSummaryWebhookTrigger meetingExternalSummaryWebhookTrigger, + SysParamService sysParamService, + RetryExecutor retryExecutor) { this.meetingMapper = meetingMapper; this.transcriptMapper = transcriptMapper; this.aiModelService = aiModelService; @@ -169,6 +197,7 @@ public class AiTaskServiceImpl extends ServiceImpl impleme this.taskSecurityContextRunner = taskSecurityContextRunner; this.meetingExternalSummaryWebhookTrigger = meetingExternalSummaryWebhookTrigger; this.sysParamService = sysParamService; + this.retryExecutor = retryExecutor; } @Override @@ -668,7 +697,17 @@ public class AiTaskServiceImpl extends ServiceImpl impleme for (int i = 0; i < 600; i++) { Thread.sleep(2000); - String queryResp = get(queryUrl, asrModel.getApiKey()); + String queryResp = retryExecutor.execute( + RetryOptions.builder() + .operation("asr-query") + .exhaustedMessage("ASR 查询超时，重试已耗尽") + .onRetry((attempt, maxAttempts, delayMs, ex) -> { + log.info("[ASR-PROC]ASR轮询结果正在重试中,meetingId={},重试次数:{},轮询url:{}", meeting.getId(), attempt, queryUrl); + updateProgress(meeting.getId(), 5, "ASR 查询超时，正在重试(" + attempt + "/" + maxAttempts + ")...", 0); + }) + .build(), + () -> get(queryUrl, asrModel.getApiKey()) + ); JsonNode statusNode = objectMapper.readTree(queryResp); int code = statusNode.path("code").asInt(500); if (code!=0){ @@ -736,7 +775,16 @@ public class AiTaskServiceImpl extends ServiceImpl impleme TaskStatus taskStatus = null; for (int i = 0; i < 600; i++) { Thread.sleep(2000); - taskStatus = queryTencentOfflineTask(asrModel, taskId); + Long currentTaskId = taskId; + taskStatus = retryExecutor.execute( + RetryOptions.builder() + .operation("tencent-asr-query") + .exhaustedMessage("腾讯离线 ASR 查询超时，重试已耗尽") + .onRetry((attempt, maxAttempts, delayMs, ex) -> + updateProgress(meeting.getId(), 5, "腾讯离线 ASR 查询超时，正在重试(" + attempt + "/" + maxAttempts + ")...", 0)) + .build(), + () -> queryTencentOfflineTask(asrModel, currentTaskId) + ); if (taskStatus == null) { throw new RuntimeException("腾讯离线 ASR 查询结果为空"); } @@ -865,8 +913,10 @@ public class AiTaskServiceImpl extends ServiceImpl impleme boolean useSpk = useSpkObj != null && useSpkObj.toString().equals("1"); config.put("enable_speaker", useSpk); config.put("match_speaker_registry", useSpk); + if (asrModel.getMediaConfig() != null) { config.put("speaker_threshold", asrModel.getMediaConfig().get("svThreshold")); - Object enableTextRefineObj = taskRecord.getTaskConfig().get("enableTextRefine"); + } + Object enableTextRefineObj = taskRecord.getTaskConfig().get("enableTextRefine"); boolean enableTextRefine = enableTextRefineObj != null && Boolean.parseBoolean(enableTextRefineObj.toString()); config.put("enable_text_cleanup", enableTextRefine); @@ -935,7 +985,18 @@ public class AiTaskServiceImpl extends ServiceImpl impleme this.updateById(taskRecord); meetingPointsService.assertSufficientPointsBeforeAsrSubmit(meeting, taskRecord); - String respBody = postJson(submitUrl, req, asrModel.getApiKey()); + String respBody = retryExecutor.execute( + RetryOptions.builder() + .operation("asr-submit") + .exhaustedMessage("ASR 提交失败，重试已耗尽") + .onRetry((attempt, maxAttempts, delayMs, ex) -> { + log.info("[ASR-PROC]ASR提交任务正在重试中,meetingId={},重试次数:{}", meeting.getId(), attempt); + updateProgress(meeting.getId(), 5, "ASR 提交失败，正在重试(" + attempt + "/" + maxAttempts + ")...", 0); + } + ) + .build(), + () -> postJson(submitUrl, req, asrModel.getApiKey()) + ); JsonNode submitNode = objectMapper.readTree(respBody); if (submitNode.path("code")==null||submitNode.path("code").asInt() != 0) { updateAiTaskFail(taskRecord, "ASR识别失败 " + respBody); diff --git a/backend/src/main/java/com/imeeting/support/retry/RetryExecutor.java b/backend/src/main/java/com/imeeting/support/retry/RetryExecutor.java new file mode 100644 index 0000000..30f9a15 --- /dev/null +++ b/backend/src/main/java/com/imeeting/support/retry/RetryExecutor.java @@ -0,0 +1,91 @@ +package com.imeeting.support.retry; + +import lombok.extern.slf4j.Slf4j; + +import java.net.ConnectException; +import java.net.http.HttpTimeoutException; +import java.util.function.Predicate; + +@Slf4j +public class RetryExecutor { + + private static final int DEFAULT_MAX_ATTEMPTS = 3; + private static final long[] DEFAULT_DELAYS_MS = new long[]{2000L, 4000L}; + + private final Sleeper sleeper; + + public RetryExecutor() { + this(Thread::sleep); + } + + public RetryExecutor(Sleeper sleeper) { + this.sleeper = sleeper; + } + + public T execute(RetryCall call) throws Exception { + return execute(null, call); + } + + public T execute(RetryOptions options, RetryCall call) throws Exception { + int maxAttempts = options == null || options.getMaxAttempts() == null ? DEFAULT_MAX_ATTEMPTS : options.getMaxAttempts(); + long[] delaysMs = options == null || options.getDelaysMs() == null || options.getDelaysMs().length == 0 + ? DEFAULT_DELAYS_MS + : options.getDelaysMs(); + Predicate retryPredicate = options == null || options.getRetryPredicate() == null + ? this::isDefaultRetryable + : options.getRetryPredicate(); + + Exception lastException = null; + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + try { + return call.execute(); + } catch (Exception ex) { + lastException = ex; + if (!retryPredicate.test(ex)) { + throw ex; + } + if (attempt >= maxAttempts) { + break; + } + long delayMs = delaysMs[Math.min(attempt - 1, delaysMs.length - 1)]; + if (options != null && options.getOnRetry() != null) { + options.getOnRetry().onRetry(attempt, maxAttempts, delayMs, ex); + } + sleeper.sleep(delayMs); + } + } + + String exhaustedMessage = options == null || options.getExhaustedMessage() == null || options.getExhaustedMessage().isBlank() + ? "重试已耗尽" + : options.getExhaustedMessage(); + throw lastException == null ? new RuntimeException(exhaustedMessage) : new RuntimeException(exhaustedMessage, lastException); + } + + protected boolean isDefaultRetryable(Throwable throwable) { + if (throwable == null) { + return false; + } + if (throwable instanceof HttpTimeoutException || throwable instanceof ConnectException) { + return true; + } + String message = throwable.getMessage(); + if (message == null || message.isBlank()) { + return false; + } + String normalized = message.toLowerCase(); + return normalized.contains("timeout") + || normalized.contains("timed out") + || normalized.contains("temporarily unavailable") + || normalized.contains("connection refused"); + } + + @FunctionalInterface + public interface RetryCall { + T execute() throws Exception; + } + + @FunctionalInterface + public interface Sleeper { + void sleep(long delayMs) throws InterruptedException; + } +} diff --git a/backend/src/main/java/com/imeeting/support/retry/RetryOptions.java b/backend/src/main/java/com/imeeting/support/retry/RetryOptions.java new file mode 100644 index 0000000..a66ffc1 --- /dev/null +++ b/backend/src/main/java/com/imeeting/support/retry/RetryOptions.java @@ -0,0 +1,23 @@ +package com.imeeting.support.retry; + +import lombok.Builder; +import lombok.Getter; + +import java.util.function.Predicate; + +@Getter +@Builder +public class RetryOptions { + + private String operation; + private Integer maxAttempts; + private long[] delaysMs; + private Predicate retryPredicate; + private RetryCallback onRetry; + private String exhaustedMessage; + + @FunctionalInterface + public interface RetryCallback { + void onRetry(int attempt, int maxAttempts, long delayMs, Exception exception); + } +} diff --git a/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java b/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java index 352332c..8e4fc12 100644 --- a/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java +++ b/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java @@ -16,6 +16,7 @@ import com.imeeting.service.biz.MeetingSummaryFileService; import com.imeeting.service.biz.MeetingTranscriptChapterService; import com.imeeting.service.biz.MeetingTranscriptFileService; import com.imeeting.support.TaskSecurityContextRunner; +import com.imeeting.support.retry.RetryExecutor; import com.imeeting.support.redis.MeetingAsrPermitCache; import com.imeeting.support.redis.MeetingLockCache; import com.unisbase.mapper.SysUserMapper; @@ -25,22 +26,215 @@ import com.tencentcloudapi.asr.v20190614.models.TaskStatus; import org.junit.jupiter.api.Test; import org.springframework.test.util.ReflectionTestUtils; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.net.ConnectException; import java.util.HashMap; import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.contains; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; class AiTaskServiceImplTest { + @Test + void processAsrTaskShouldRetryOrdinaryOfflineQueryWhenTimeoutOccurs() throws Exception { + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + AiTaskServiceImpl service = spy(createService(meetingPointsService)); + + HttpClient httpClient = mock(HttpClient.class); + @SuppressWarnings("unchecked") + HttpResponse submitResponse = mock(HttpResponse.class); + @SuppressWarnings("unchecked") + HttpResponse queryResponse = mock(HttpResponse.class); + ReflectionTestUtils.setField(service, "httpClient", httpClient); + ReflectionTestUtils.setField(service, "serverBaseUrl", "https://server.example.com"); + + Meeting meeting = new Meeting(); + meeting.setId(100L); + meeting.setAudioUrl("/upload/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setId(1001L); + task.setMeetingId(100L); + task.setTaskType("ASR"); + task.setTaskConfig(new HashMap<>(Map.of( + "asrModelId", 501L, + "useSpkId", 0, + "enableTextRefine", true + ))); + + AiModelVO model = new AiModelVO(); + model.setId(501L); + model.setProvider("local"); + model.setBaseUrl("https://asr.example.com"); + model.setApiKey("api-key"); + model.setMediaConfig(Map.of("svThreshold", 0.5D)); + when(extractAiModelService(service).getModelById(501L, "ASR")).thenReturn(model); + + when(submitResponse.body()).thenReturn("{\"code\":0,\"data\":{\"task_id\":\"task-1001\"}}"); + when(queryResponse.body()).thenReturn("{\"code\":0,\"data\":{\"status\":\"completed\",\"result\":{\"segments\":[{\"speaker_id\":\"spk_1\",\"speaker_name\":\"张三\",\"text\":\"测试转写\"}]}}}"); + when(httpClient.send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class))) + .thenReturn(submitResponse) + .thenThrow(new HttpTimeoutException("request timed out")) + .thenThrow(new HttpTimeoutException("request timed out")) + .thenReturn(queryResponse); + doReturn(true).when(service).updateById(any(AiTask.class)); + + String result = ReflectionTestUtils.invokeMethod(service, "processAsrTask", meeting, task); + + assertEquals("张三: 测试转写\n", result); + verify(httpClient, times(4)).send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class)); + } + + @Test + void processTencentOfflineAsrShouldFailAfterTencentQueryRetryExhausted() throws Exception { + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + AiTaskServiceImpl service = spy(createService(meetingPointsService)); + + Meeting meeting = new Meeting(); + meeting.setId(200L); + meeting.setAudioUrl("https://cdn.example.com/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setId(2001L); + task.setMeetingId(200L); + task.setTaskType("ASR"); + task.setTaskConfig(new HashMap<>(Map.of( + "asrModelId", 601L, + "useSpkId", 1 + ))); + + AiModelVO model = new AiModelVO(); + model.setProvider("tencent"); + model.setModelCode("legacy-model-code"); + model.setMediaConfig(Map.of( + "tencentAppId", "123456", + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key", + "tencentOfflineModelCode", "16k_zh" + )); + + doReturn(90001L).when(service).submitTencentOfflineTask(meeting, task, model); + doThrow(new com.tencentcloudapi.common.exception.TencentCloudSDKException("Request timeout")) + .when(service).queryTencentOfflineTask(model, 90001L); + + RuntimeException ex = assertThrows(RuntimeException.class, () -> service.processTencentOfflineAsr(meeting, task, model)); + + assertTrue(ex.getMessage().contains("查询超时")); + verify(service, times(3)).queryTencentOfflineTask(model, 90001L); + } + + @Test + void isRetryableAsrQueryExceptionShouldTreatHttpTimeoutAsRetryable() { + AiTaskServiceImpl service = createService(mock(MeetingPointsService.class)); + + assertTrue(service.isRetryableAsrQueryException(new HttpTimeoutException("request timed out"))); + } + + @Test + void processAsrTaskShouldRetryOrdinaryOfflineSubmitWhenConnectExceptionOccurs() throws Exception { + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + AiTaskServiceImpl service = spy(createService(meetingPointsService)); + + HttpClient httpClient = mock(HttpClient.class); + @SuppressWarnings("unchecked") + HttpResponse submitResponse = mock(HttpResponse.class); + @SuppressWarnings("unchecked") + HttpResponse queryResponse = mock(HttpResponse.class); + ReflectionTestUtils.setField(service, "httpClient", httpClient); + ReflectionTestUtils.setField(service, "serverBaseUrl", "https://server.example.com"); + + Meeting meeting = new Meeting(); + meeting.setId(300L); + meeting.setAudioUrl("/upload/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setId(3001L); + task.setMeetingId(300L); + task.setTaskType("ASR"); + task.setTaskConfig(new HashMap<>(Map.of( + "asrModelId", 701L, + "useSpkId", 0, + "enableTextRefine", true + ))); + + AiModelVO model = new AiModelVO(); + model.setId(701L); + model.setProvider("local"); + model.setBaseUrl("https://asr.example.com"); + model.setApiKey("api-key"); + model.setMediaConfig(Map.of("svThreshold", 0.5D)); + when(extractAiModelService(service).getModelById(701L, "ASR")).thenReturn(model); + + when(submitResponse.body()).thenReturn("{\"code\":0,\"data\":{\"task_id\":\"task-3001\"}}"); + when(queryResponse.body()).thenReturn("{\"code\":0,\"data\":{\"status\":\"completed\",\"result\":{\"segments\":[{\"speaker_id\":\"spk_1\",\"speaker_name\":\"李四\",\"text\":\"提交重试成功\"}]}}}"); + when(httpClient.send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class))) + .thenThrow(new ConnectException("Connection refused")) + .thenReturn(submitResponse) + .thenReturn(queryResponse); + doReturn(true).when(service).updateById(any(AiTask.class)); + + String result = ReflectionTestUtils.invokeMethod(service, "processAsrTask", meeting, task); + + assertEquals("李四: 提交重试成功\n", result); + verify(httpClient, times(3)).send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class)); + } + + @Test + void processAsrTaskShouldNotRetryOrdinaryOfflineSubmitWhenTimeoutOccurs() throws Exception { + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + AiTaskServiceImpl service = spy(createService(meetingPointsService)); + + HttpClient httpClient = mock(HttpClient.class); + ReflectionTestUtils.setField(service, "httpClient", httpClient); + ReflectionTestUtils.setField(service, "serverBaseUrl", "https://server.example.com"); + + Meeting meeting = new Meeting(); + meeting.setId(400L); + meeting.setAudioUrl("/upload/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setId(4001L); + task.setMeetingId(400L); + task.setTaskType("ASR"); + task.setTaskConfig(new HashMap<>(Map.of( + "asrModelId", 801L, + "useSpkId", 0, + "enableTextRefine", true + ))); + + AiModelVO model = new AiModelVO(); + model.setId(801L); + model.setProvider("local"); + model.setBaseUrl("https://asr.example.com"); + model.setApiKey("api-key"); + model.setMediaConfig(Map.of("svThreshold", 0.5D)); + when(extractAiModelService(service).getModelById(801L, "ASR")).thenReturn(model); + + when(httpClient.send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class))) + .thenThrow(new HttpTimeoutException("request timed out")); + doReturn(true).when(service).updateById(any(AiTask.class)); + + assertThrows(Exception.class, () -> ReflectionTestUtils.invokeMethod(service, "processAsrTask", meeting, task)); + verify(httpClient, times(1)).send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class)); + } + @Test void processAsrTaskShouldUseTencentOfflineBranchWhenProviderIsTencent() throws Exception { MeetingMapper meetingMapper = mock(MeetingMapper.class); @@ -76,7 +270,9 @@ class AiTaskServiceImplTest { meetingSummaryPromptAssembler, taskSecurityContextRunner, meetingExternalSummaryWebhookTrigger, - sysParamService + sysParamService, + new RetryExecutor(delayMs -> { + }) )); ReflectionTestUtils.setField(service, "baseMapper", mock(AiTaskMapper.class)); ReflectionTestUtils.setField(service, "androidMeetingPushService", mock(AndroidMeetingPushService.class)); @@ -299,10 +495,16 @@ class AiTaskServiceImplTest { mock(MeetingSummaryPromptAssembler.class), mock(TaskSecurityContextRunner.class), mock(MeetingExternalSummaryWebhookTrigger.class), - mock(SysParamService.class) + mock(SysParamService.class), + new RetryExecutor(delayMs -> { + }) ); ReflectionTestUtils.setField(service, "baseMapper", mock(AiTaskMapper.class)); ReflectionTestUtils.setField(service, "androidMeetingPushService", mock(AndroidMeetingPushService.class)); return service; } + + private AiModelService extractAiModelService(AiTaskServiceImpl service) { + return (AiModelService) ReflectionTestUtils.getField(service, "aiModelService"); + } }