From 83432cf1c44d303f165e6e036073a26644dd3c77 Mon Sep 17 00:00:00 2001 From: codychen123 Date: Sun, 10 May 2026 16:26:35 +0800 Subject: [PATCH] fix: run leader research asynchronously --- .../research/LeaderResearchController.kt | 6 +- .../research/LeaderResearchJobService.kt | 126 ++++++++++++------ .../research/LeaderResearchControllerTest.kt | 37 ++++- .../research/LeaderResearchJobServiceTest.kt | 21 +++ frontend/src/pages/LeaderResearch.tsx | 22 ++- 5 files changed, 164 insertions(+), 48 deletions(-) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchController.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchController.kt index 8cd672c..13c330b 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchController.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchController.kt @@ -51,7 +51,11 @@ class LeaderResearchController( return try { val trigger = runCatching { LeaderResearchTriggerType.valueOf(request.triggerType.uppercase()) } .getOrDefault(LeaderResearchTriggerType.MANUAL) - val run = jobService.runOnce(request.dryRun, trigger) + val run = if (request.dryRun || trigger == LeaderResearchTriggerType.PREVIEW) { + jobService.runOnce(request.dryRun, trigger) + } else { + jobService.startAsync(request.dryRun, trigger) + } ResponseEntity.ok(ApiResponse.success(mapper.runDto(run))) } catch (e: Exception) { logger.error("Leader research run failed", e) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobService.kt index 25b4406..a18b08a 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobService.kt @@ -9,11 +9,13 @@ import com.wrbug.polymarketbot.enums.LeaderResearchTriggerType import com.wrbug.polymarketbot.repository.LeaderActivityEventRepository import com.wrbug.polymarketbot.repository.LeaderResearchCandidateRepository import com.wrbug.polymarketbot.repository.LeaderResearchRunRepository +import jakarta.annotation.PreDestroy import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Value import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Service -import org.springframework.transaction.annotation.Transactional +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors import java.util.concurrent.atomic.AtomicBoolean @Service @@ -30,6 +32,9 @@ class LeaderResearchJobService( ) { private val logger = LoggerFactory.getLogger(LeaderResearchJobService::class.java) private val running = AtomicBoolean(false) + internal var runExecutor: ExecutorService = Executors.newSingleThreadExecutor { runnable -> + Thread(runnable, "leader-research-runner").apply { isDaemon = true } + } @Scheduled(fixedDelayString = "\${leader.research.fixed-delay-ms:900000}") fun scheduledRun() { @@ -37,8 +42,37 @@ class LeaderResearchJobService( runOnce(dryRun = false, triggerType = LeaderResearchTriggerType.SCHEDULED) } - @Transactional fun runOnce(dryRun: Boolean, triggerType: LeaderResearchTriggerType = LeaderResearchTriggerType.MANUAL): LeaderResearchRun { + val start = startRunRecord(dryRun, triggerType) + if (!start.acquired) return start.run + return executeStartedRun(start.run) + } + + fun startAsync(dryRun: Boolean, triggerType: LeaderResearchTriggerType = LeaderResearchTriggerType.MANUAL): LeaderResearchRun { + val start = startRunRecord(dryRun, triggerType) + if (!start.acquired) return start.run + + return try { + runExecutor.submit { + logger.info("Leader research async run started: runId={}", start.run.id) + executeStartedRun(start.run) + } + logger.info("Leader research async run queued: runId={}, triggerType={}", start.run.id, triggerType) + start.run + } catch (e: RuntimeException) { + logger.error("Failed to queue leader research async run: runId={}", start.run.id, e) + failStartedRun(start.run, e).also { running.set(false) } + } + } + + @PreDestroy + fun shutdown() { + runExecutor.shutdownNow() + } + + private data class RunStart(val run: LeaderResearchRun, val acquired: Boolean) + + private fun startRunRecord(dryRun: Boolean, triggerType: LeaderResearchTriggerType): RunStart { if (!running.compareAndSet(false, true)) { val now = System.currentTimeMillis() val skipped = runRepository.save( @@ -59,28 +93,38 @@ class LeaderResearchJobService( runId = skipped.id, reason = "Skipped because another research run is in progress" ) - return skipped + return RunStart(skipped, acquired = false) } val startedAt = System.currentTimeMillis() - var run = runRepository.save( - LeaderResearchRun( - status = LeaderResearchRunStatus.RUNNING, - triggerType = triggerType, - dryRun = dryRun, - startedAt = startedAt, - createdAt = startedAt, - updatedAt = startedAt - ) - ) - eventService.record( - type = LeaderResearchEventType.RUN_STARTED, - runId = run.id, - reason = "Leader research run started" - ) - return try { - val isPreview = dryRun || triggerType == LeaderResearchTriggerType.PREVIEW + val run = runRepository.save( + LeaderResearchRun( + status = LeaderResearchRunStatus.RUNNING, + triggerType = triggerType, + dryRun = dryRun, + startedAt = startedAt, + createdAt = startedAt, + updatedAt = startedAt + ) + ) + eventService.record( + type = LeaderResearchEventType.RUN_STARTED, + runId = run.id, + reason = "Leader research run started" + ) + RunStart(run, acquired = true) + } catch (e: RuntimeException) { + running.set(false) + throw e + } + } + + private fun executeStartedRun(startedRun: LeaderResearchRun): LeaderResearchRun { + var run = startedRun + val startedAt = startedRun.startedAt + return try { + val isPreview = run.dryRun || run.triggerType == LeaderResearchTriggerType.PREVIEW val sourceResults = if (isPreview) sourceService.previewCandidates() else sourceService.discoverCandidates(run.id) if (!isPreview) { scoringService.scoreAll(run.id) @@ -121,27 +165,31 @@ class LeaderResearchJobService( ) run } catch (e: Exception) { - logger.error("Leader research run failed", e) - val now = System.currentTimeMillis() - runRepository.save( - run.copy( - status = LeaderResearchRunStatus.FAILED, - finishedAt = now, - durationMs = now - startedAt, - errorClass = e::class.java.simpleName, - errorMessage = e.message, - updatedAt = now - ) - ).also { - eventService.record( - type = LeaderResearchEventType.RUN_FAILED, - runId = it.id, - reason = e.message, - payloadSummary = e::class.java.name - ) - } + failStartedRun(run, e) } finally { running.set(false) } } + + private fun failStartedRun(run: LeaderResearchRun, e: Exception): LeaderResearchRun { + logger.error("Leader research run failed", e) + val now = System.currentTimeMillis() + return runRepository.save( + run.copy( + status = LeaderResearchRunStatus.FAILED, + finishedAt = now, + durationMs = now - run.startedAt, + errorClass = e::class.java.simpleName, + errorMessage = e.message, + updatedAt = now + ) + ).also { + eventService.record( + type = LeaderResearchEventType.RUN_FAILED, + runId = it.id, + reason = e.message, + payloadSummary = e::class.java.name + ) + } + } } diff --git a/backend/src/test/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchControllerTest.kt b/backend/src/test/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchControllerTest.kt index 70a8d5a..adc9b5e 100644 --- a/backend/src/test/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchControllerTest.kt +++ b/backend/src/test/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchControllerTest.kt @@ -3,6 +3,7 @@ package com.wrbug.polymarketbot.controller.copytrading.research import com.wrbug.polymarketbot.dto.LeaderResearchApprovalRequest import com.wrbug.polymarketbot.dto.LeaderResearchRunRequest import com.wrbug.polymarketbot.entity.LeaderResearchRun +import com.wrbug.polymarketbot.enums.LeaderResearchTriggerType import com.wrbug.polymarketbot.enums.ErrorCode import com.wrbug.polymarketbot.service.copytrading.research.LeaderResearchApprovalConfirmRequiredException import com.wrbug.polymarketbot.service.copytrading.research.LeaderResearchApprovalService @@ -29,9 +30,9 @@ class LeaderResearchControllerTest { ) @Test - fun `run returns run dto`() { + fun `manual run queues async run and returns run dto`() { val run = LeaderResearchRun(id = 1L) - Mockito.`when`(jobService.runOnce(false, com.wrbug.polymarketbot.enums.LeaderResearchTriggerType.MANUAL)).thenReturn(run) + Mockito.`when`(jobService.startAsync(false, LeaderResearchTriggerType.MANUAL)).thenReturn(run) Mockito.`when`(mapper.runDto(run)).thenReturn( com.wrbug.polymarketbot.dto.LeaderResearchRunDto( id = 1, @@ -54,6 +55,38 @@ class LeaderResearchControllerTest { assertEquals(0, response.body!!.code) assertEquals(1, response.body!!.data!!.id) + Mockito.verify(jobService).startAsync(false, LeaderResearchTriggerType.MANUAL) + Mockito.verify(jobService, Mockito.never()).runOnce(false, LeaderResearchTriggerType.MANUAL) + } + + @Test + fun `preview run stays synchronous`() { + val run = LeaderResearchRun(id = 2L, dryRun = true, triggerType = LeaderResearchTriggerType.PREVIEW) + Mockito.`when`(jobService.runOnce(true, LeaderResearchTriggerType.PREVIEW)).thenReturn(run) + Mockito.`when`(mapper.runDto(run)).thenReturn( + com.wrbug.polymarketbot.dto.LeaderResearchRunDto( + id = 2, + status = "SUCCESS", + triggerType = "PREVIEW", + dryRun = true, + startedAt = run.startedAt, + finishedAt = null, + durationMs = null, + sourceCountsJson = null, + candidateCountsJson = null, + partialFailure = false, + skippedReason = null, + errorClass = null, + errorMessage = null + ) + ) + + val response = controller.run(LeaderResearchRunRequest(dryRun = true, triggerType = "PREVIEW")) + + assertEquals(0, response.body!!.code) + assertEquals(2, response.body!!.data!!.id) + Mockito.verify(jobService).runOnce(true, LeaderResearchTriggerType.PREVIEW) + Mockito.verify(jobService, Mockito.never()).startAsync(true, LeaderResearchTriggerType.PREVIEW) } @Test diff --git a/backend/src/test/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobServiceTest.kt b/backend/src/test/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobServiceTest.kt index d218cbe..de34e95 100644 --- a/backend/src/test/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobServiceTest.kt +++ b/backend/src/test/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobServiceTest.kt @@ -16,6 +16,8 @@ import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test import org.mockito.Mockito +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit class LeaderResearchJobServiceTest { private val runRepository: LeaderResearchRunRepository = mock() @@ -131,6 +133,25 @@ class LeaderResearchJobServiceTest { Mockito.verifyNoInteractions(scoringService, stateMachine, paperTradingService) } + @Test + fun `async run returns running record before background execution completes`() { + val service = service() + val executor = Executors.newSingleThreadExecutor() + service.runExecutor = executor + stubRunSaves() + Mockito.`when`(sourceService.discoverCandidates(1L)).thenAnswer { + Thread.sleep(100) + emptyList() + } + + val run = service.startAsync(dryRun = false, triggerType = LeaderResearchTriggerType.MANUAL) + + assertEquals(LeaderResearchRunStatus.RUNNING, run.status) + executor.shutdown() + assertTrue(executor.awaitTermination(2, TimeUnit.SECONDS)) + Mockito.verify(sourceService).discoverCandidates(run.id) + } + @Test fun `overlap guard records skipped run while outer run continues`() { lateinit var service: LeaderResearchJobService diff --git a/frontend/src/pages/LeaderResearch.tsx b/frontend/src/pages/LeaderResearch.tsx index 6e70a58..8606ecc 100644 --- a/frontend/src/pages/LeaderResearch.tsx +++ b/frontend/src/pages/LeaderResearch.tsx @@ -99,8 +99,8 @@ const LeaderResearch: React.FC = () => { const [approvalLoading, setApprovalLoading] = useState(false) const [approvalForm] = Form.useForm() - const loadAll = async () => { - setLoading(true) + const loadAll = async (showLoading = true) => { + if (showLoading) setLoading(true) try { const [candidateResp, summaryResp, sourceResp, accountResp] = await Promise.all([ apiService.leaderResearch.listCandidates({ page: 0, size: 50, state: stateFilter, query: query || undefined }), @@ -125,7 +125,7 @@ const LeaderResearch: React.FC = () => { } catch (error: any) { message.error(error.message || t('leaderResearch.fetchFailed')) } finally { - setLoading(false) + if (showLoading) setLoading(false) } } @@ -133,6 +133,16 @@ const LeaderResearch: React.FC = () => { loadAll() }, [stateFilter]) + useEffect(() => { + const lastRunStatus = summary?.lastRun?.status || candidates.summary?.lastRun?.status + if (lastRunStatus !== 'RUNNING') return + + const timer = window.setInterval(() => { + loadAll(false) + }, 5000) + return () => window.clearInterval(timer) + }, [summary?.lastRun?.status, candidates.summary?.lastRun?.status]) + const runAgent = async () => { setRunning(true) try { @@ -291,8 +301,8 @@ const LeaderResearch: React.FC = () => { {t('leaderResearch.subtitle')} - - + @@ -390,7 +400,7 @@ const LeaderResearch: React.FC = () => { placeholder={t('leaderResearch.searchPlaceholder')} value={query} onChange={event => setQuery(event.target.value)} - onSearch={loadAll} + onSearch={() => loadAll()} />