feat: resume abandoned sessions on server restart
Three coordinated changes: 1. DefaultOrchestrationReducer now handles TransitionExecutedEvent, updating currentStageId so the projected state always reflects the actual current stage rather than staying stuck at the start stage. 2. DefaultSessionOrchestrator.resume() re-enters the execution loop at the current stage without re-emitting WorkflowStartedEvent. It reads currentStageId from the projected state, creates an ExecutionContext there, and calls enterStage → step exactly as run() does after its first stage. 3. ServerModule.start() calls resumeAbandonedSessions(), which scans all RUNNING/PAUSED sessions on startup and launches resume() for each one whose orchestrator is no longer in memory. Sessions that have a live deferred are unaffected; only sessions with no coroutine (e.g. the server was restarted while a tool was executing or approval was pending) are picked up and re-executed from their current stage.
This commit is contained in:
@@ -16,13 +16,18 @@ import com.correx.core.kernel.orchestration.OrchestrationRepository
|
|||||||
import com.correx.core.router.RouterFacade
|
import com.correx.core.router.RouterFacade
|
||||||
import com.correx.core.sessions.DefaultSessionRepository
|
import com.correx.core.sessions.DefaultSessionRepository
|
||||||
import com.correx.core.tools.registry.ToolRegistry
|
import com.correx.core.tools.registry.ToolRegistry
|
||||||
|
import com.correx.core.events.orchestration.OrchestrationStatus
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
import kotlinx.coroutines.Job
|
import kotlinx.coroutines.Job
|
||||||
import kotlinx.coroutines.SupervisorJob
|
import kotlinx.coroutines.SupervisorJob
|
||||||
import kotlinx.coroutines.flow.filter
|
import kotlinx.coroutines.flow.filter
|
||||||
import kotlinx.coroutines.flow.onEach
|
|
||||||
import kotlinx.coroutines.flow.launchIn
|
import kotlinx.coroutines.flow.launchIn
|
||||||
|
import kotlinx.coroutines.flow.onEach
|
||||||
|
import kotlinx.coroutines.launch
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
|
|
||||||
|
private val log = LoggerFactory.getLogger(ServerModule::class.java)
|
||||||
|
|
||||||
class ServerModule(
|
class ServerModule(
|
||||||
val orchestrator: DefaultSessionOrchestrator,
|
val orchestrator: DefaultSessionOrchestrator,
|
||||||
@@ -63,12 +68,34 @@ class ServerModule(
|
|||||||
fun start() {
|
fun start() {
|
||||||
if (subscriptionJob != null) return
|
if (subscriptionJob != null) return
|
||||||
preRegisterPendingApprovals()
|
preRegisterPendingApprovals()
|
||||||
|
resumeAbandonedSessions()
|
||||||
subscriptionJob = eventStore.subscribeAll()
|
subscriptionJob = eventStore.subscribeAll()
|
||||||
.filter { it.payload is ApprovalRequestedEvent }
|
.filter { it.payload is ApprovalRequestedEvent }
|
||||||
.onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) }
|
.onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) }
|
||||||
.launchIn(moduleScope)
|
.launchIn(moduleScope)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun resumeAbandonedSessions() {
|
||||||
|
eventStore.allSessionIds().forEach { sessionId ->
|
||||||
|
val orchState = orchestrationRepository.getState(sessionId)
|
||||||
|
if (orchState.status != OrchestrationStatus.RUNNING &&
|
||||||
|
orchState.status != OrchestrationStatus.PAUSED
|
||||||
|
) return@forEach
|
||||||
|
val graph = workflowRegistry.find(orchState.workflowId) ?: run {
|
||||||
|
log.warn("resumeAbandoned: no graph for session={} workflowId={}", sessionId.value, orchState.workflowId)
|
||||||
|
return@forEach
|
||||||
|
}
|
||||||
|
log.info("resumeAbandoned: launching session={} workflow={}", sessionId.value, orchState.workflowId)
|
||||||
|
moduleScope.launch {
|
||||||
|
runCatching {
|
||||||
|
orchestrator.resume(sessionId, graph, defaultOrchestrationConfig)
|
||||||
|
}.onFailure { e ->
|
||||||
|
log.error("resumeAbandoned: session={} failed", sessionId.value, e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private fun preRegisterPendingApprovals() {
|
private fun preRegisterPendingApprovals() {
|
||||||
val projector = ApprovalProjector(DefaultApprovalReducer())
|
val projector = ApprovalProjector(DefaultApprovalReducer())
|
||||||
eventStore.allSessionIds().forEach { sessionId ->
|
eventStore.allSessionIds().forEach { sessionId ->
|
||||||
|
|||||||
+3
@@ -4,6 +4,7 @@ import com.correx.core.events.events.OrchestrationPausedEvent
|
|||||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||||
import com.correx.core.events.events.RetryAttemptedEvent
|
import com.correx.core.events.events.RetryAttemptedEvent
|
||||||
import com.correx.core.events.events.StoredEvent
|
import com.correx.core.events.events.StoredEvent
|
||||||
|
import com.correx.core.events.events.TransitionExecutedEvent
|
||||||
import com.correx.core.events.events.WorkflowCompletedEvent
|
import com.correx.core.events.events.WorkflowCompletedEvent
|
||||||
import com.correx.core.events.events.WorkflowFailedEvent
|
import com.correx.core.events.events.WorkflowFailedEvent
|
||||||
import com.correx.core.events.events.WorkflowStartedEvent
|
import com.correx.core.events.events.WorkflowStartedEvent
|
||||||
@@ -44,6 +45,8 @@ class DefaultOrchestrationReducer : OrchestrationReducer {
|
|||||||
pauseReason = null,
|
pauseReason = null,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
is TransitionExecutedEvent -> state.copy(currentStageId = p.to)
|
||||||
|
|
||||||
is RetryAttemptedEvent -> state.copy(
|
is RetryAttemptedEvent -> state.copy(
|
||||||
status = OrchestrationStatus.RUNNING,
|
status = OrchestrationStatus.RUNNING,
|
||||||
retryCount = p.attemptNumber,
|
retryCount = p.attemptNumber,
|
||||||
|
|||||||
+16
@@ -136,6 +136,22 @@ class DefaultSessionOrchestrator(
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
suspend fun resume(
|
||||||
|
sessionId: SessionId,
|
||||||
|
graph: WorkflowGraph,
|
||||||
|
config: OrchestrationConfig,
|
||||||
|
): WorkflowResult {
|
||||||
|
val stageId = orchestrationRepository.getState(sessionId).currentStageId
|
||||||
|
?: return WorkflowResult.Failed(sessionId, "resume: no currentStageId", retryExhausted = false)
|
||||||
|
log.info("[Orchestrator] resuming session={} workflow={} stage={}", sessionId.value, graph.id, stageId.value)
|
||||||
|
val base = ExecutionContext(graph, sessionId, 0, stageId, config, null, null)
|
||||||
|
val enriched = base.enrich()
|
||||||
|
return when (val result = enterStage(enriched, stageId)) {
|
||||||
|
is StepResult.Continue -> step(result.ctx.copy(currentStageId = stageId))
|
||||||
|
is StepResult.Terminal -> result.result
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
override suspend fun cancel(sessionId: SessionId) {
|
override suspend fun cancel(sessionId: SessionId) {
|
||||||
cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true)
|
cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user