gavindiaz/PolyHermes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: run leader research asynchronously

Browse Source
This commit is contained in:
codychen123
2026-05-10 16:26:35 +08:00
parent 017c58d5bf
commit 83432cf1c4
5 changed files with 164 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchController.kt
+5 -1
View File
@@ -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)
backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobService.kt
+87 -39
View File
@@ -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
)
}
}
}
backend/src/test/kotlin/com/wrbug/polymarketbot/controller/copytrading/research/LeaderResearchControllerTest.kt
+35 -2
View File
@@ -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
backend/src/test/kotlin/com/wrbug/polymarketbot/service/copytrading/research/LeaderResearchJobServiceTest.kt
+21
View File
@@ -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<LeaderResearchSourceRunResult>()
}
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
frontend/src/pages/LeaderResearch.tsx
+16 -6
View File
@@ -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 = () => {
<Paragraph type="secondary" style={{ marginBottom: 0 }}>{t('leaderResearch.subtitle')}</Paragraph>
</div>
<Space>
<Button icon={<ReloadOutlined />} onClick={loadAll}>{t('common.refresh')}</Button>
<Button type="primary" icon={<PlayCircleOutlined />} loading={running} onClick={runAgent}>
<Button icon={<ReloadOutlined />} onClick={() => loadAll()}>{t('common.refresh')}</Button>
<Button type="primary" icon={<PlayCircleOutlined />} loading={running || lastRun?.status === 'RUNNING'} onClick={runAgent}>
{t('leaderResearch.runNow')}
</Button>
</Space>
@@ -390,7 +400,7 @@ const LeaderResearch: React.FC = () => {
placeholder={t('leaderResearch.searchPlaceholder')}
value={query}
onChange={event => setQuery(event.target.value)}
onSearch={loadAll}
onSearch={() => loadAll()}
/>
</Space>
<Table