fix(server): record boot-time parking of stale sessions as OrchestrationPaused
A stale RUNNING session skipped by resumeAbandoned stayed RUNNING in every projection — the TUI showed a phantom running workflow that no process was executing, and each boot re-logged the same skip. Parking now emits OrchestrationPaused(reason=ABANDONED_STALE): status becomes PAUSED (truthful — not executing, resumable via POST /resume) and subsequent boots skip it on status alone.
This commit is contained in:
@@ -18,11 +18,13 @@ import com.correx.core.events.events.NewEvent
|
|||||||
import com.correx.core.events.events.ExecutionPlanLockedEvent
|
import com.correx.core.events.events.ExecutionPlanLockedEvent
|
||||||
import com.correx.core.events.events.OperatorProfileBoundEvent
|
import com.correx.core.events.events.OperatorProfileBoundEvent
|
||||||
import com.correx.core.events.events.ProjectProfileBoundEvent
|
import com.correx.core.events.events.ProjectProfileBoundEvent
|
||||||
|
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.WorkflowFailedEvent
|
import com.correx.core.events.events.WorkflowFailedEvent
|
||||||
import com.correx.core.events.stores.EventStore
|
import com.correx.core.events.stores.EventStore
|
||||||
import com.correx.core.events.types.EventId
|
import com.correx.core.events.types.EventId
|
||||||
import com.correx.core.events.types.SessionId
|
import com.correx.core.events.types.SessionId
|
||||||
|
import com.correx.core.events.types.StageId
|
||||||
import com.correx.core.kernel.execution.WorkflowResult
|
import com.correx.core.kernel.execution.WorkflowResult
|
||||||
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
|
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
|
||||||
import com.correx.core.kernel.orchestration.OrchestrationConfig
|
import com.correx.core.kernel.orchestration.OrchestrationConfig
|
||||||
@@ -366,9 +368,31 @@ class ServerModule(
|
|||||||
val lastEventAt = eventStore.read(sessionId).lastOrNull()?.metadata?.timestamp ?: return@forEach
|
val lastEventAt = eventStore.read(sessionId).lastOrNull()?.metadata?.timestamp ?: return@forEach
|
||||||
if (lastEventAt < cutoff) {
|
if (lastEventAt < cutoff) {
|
||||||
log.info(
|
log.info(
|
||||||
"resumeAbandoned: skipping stale session={} (last event {}, cutoff {}m) — POST /resume revives it",
|
"resumeAbandoned: parking stale session={} (last event {}, cutoff {}m) — POST /resume revives it",
|
||||||
sessionId.value, lastEventAt, maxAgeMinutes,
|
sessionId.value, lastEventAt, maxAgeMinutes,
|
||||||
)
|
)
|
||||||
|
// Record the parking decision as an event (invariant #9): the projection
|
||||||
|
// otherwise reports RUNNING forever for a session no process is executing,
|
||||||
|
// and the next boot would re-log the same skip.
|
||||||
|
moduleScope.launch {
|
||||||
|
eventStore.append(
|
||||||
|
NewEvent(
|
||||||
|
metadata = EventMetadata(
|
||||||
|
eventId = EventId(java.util.UUID.randomUUID().toString()),
|
||||||
|
sessionId = sessionId,
|
||||||
|
timestamp = Clock.System.now(),
|
||||||
|
schemaVersion = 1,
|
||||||
|
causationId = null,
|
||||||
|
correlationId = null,
|
||||||
|
),
|
||||||
|
payload = OrchestrationPausedEvent(
|
||||||
|
sessionId = sessionId,
|
||||||
|
stageId = orchState.currentStageId ?: StageId.NONE,
|
||||||
|
reason = "ABANDONED_STALE",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
return@forEach
|
return@forEach
|
||||||
}
|
}
|
||||||
val graph = workflowRegistry.find(orchState.workflowId) ?: run {
|
val graph = workflowRegistry.find(orchState.workflowId) ?: run {
|
||||||
|
|||||||
Reference in New Issue
Block a user