feat: 添加重试机制以增强 ASR 任务的稳定性

- 在 `AiTaskServiceImpl` 中引入 `RetryExecutor` 和 `RetryOptions`，以处理 ASR 任务提交和查询的超时及连接异常
- 添加多个单元测试以验证 ASR 任务重试逻辑

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<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> impleme
         this.taskSecurityContextRunner = taskSecurityContextRunner;
         this.meetingExternalSummaryWebhookTrigger = meetingExternalSummaryWebhookTrigger;
         this.sysParamService = sysParamService;
+    this.retryExecutor = retryExecutor;
     }
 
     @Override
@@ -668,7 +697,17 @@ public class AiTaskServiceImpl extends ServiceImpl<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> 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);

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> T execute(RetryCall<T> call) throws Exception {
+    return execute(null, call);
+  }
+
+  public <T> T execute(RetryOptions options, RetryCall<T> 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<Throwable> 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> {
+    T execute() throws Exception;
+  }
+
+  @FunctionalInterface
+  public interface Sleeper {
+    void sleep(long delayMs) throws InterruptedException;
+  }
+}

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<Throwable> retryPredicate;
+  private RetryCallback onRetry;
+  private String exhaustedMessage;
+
+  @FunctionalInterface
+  public interface RetryCallback {
+    void onRetry(int attempt, int maxAttempts, long delayMs, Exception exception);
+  }
+}

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<String> submitResponse = mock(HttpResponse.class);
+    @SuppressWarnings("unchecked")
+    HttpResponse<String> 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<String> submitResponse = mock(HttpResponse.class);
+    @SuppressWarnings("unchecked")
+    HttpResponse<String> 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");
+  }
 }