Add a wait on stop run and remove running check on delete

run/src/main/kotlin/com/cosmotech/run/service/RunServiceImpl.kt

@@ -333,11 +333,10 @@ class RunServiceImpl(
         getRun(runner.organizationId, runner.workspaceId, runner.id, runner.lastRunInfo.lastRunId!!)
     run.hasPermission(PERMISSION_WRITE)
 
-    if (!(run.state!!.isTerminal())) {
+    check(!(run.state!!.isTerminal())) {
       logger.warn(
           "Run ${run.id} is already in a terminal state (${run.state}). It can't be stopped."
       )
-      return // exiting, run already stopped
     }
 
     workflowService.stopWorkflow(run)

run/src/test/kotlin/com/cosmotech/run/service/RunServiceImplTests.kt

@@ -0,0 +1,147 @@
+// Copyright (c) Cosmo Tech.
+// Licensed under the MIT license.
+package com.cosmotech.run.service
+
+import com.cosmotech.common.events.RunStop
+import com.cosmotech.common.rbac.CsmRbac
+import com.cosmotech.common.rbac.ROLE_ADMIN
+import com.cosmotech.run.RunContainerFactory
+import com.cosmotech.run.domain.Run
+import com.cosmotech.run.domain.RunEditInfo
+import com.cosmotech.run.domain.RunState
+import com.cosmotech.run.domain.RunStatus
+import com.cosmotech.run.repository.RunRepository
+import com.cosmotech.run.workflow.WorkflowService
+import com.cosmotech.runner.RunnerApiServiceInterface
+import com.cosmotech.runner.domain.LastRunInfo
+import com.cosmotech.runner.domain.Runner
+import com.cosmotech.runner.domain.RunnerAccessControl
+import com.cosmotech.runner.domain.RunnerDatasets
+import com.cosmotech.runner.domain.RunnerEditInfo
+import com.cosmotech.runner.domain.RunnerSecurity
+import com.cosmotech.runner.domain.RunnerValidationStatus
+import io.mockk.MockKAnnotations
+import io.mockk.every
+import io.mockk.impl.annotations.InjectMockKs
+import io.mockk.impl.annotations.MockK
+import io.mockk.impl.annotations.RelaxedMockK
+import io.mockk.junit5.MockKExtension
+import io.mockk.slot
+import io.mockk.verify
+import java.util.Optional
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.extension.ExtendWith
+
+private const val ORGANIZATION_ID = "o-organizationid"
+private const val WORKSPACE_ID = "w-workspaceid"
+private const val RUNNER_ID = "r-runnerid"
+private const val RUN_ID = "run-runid"
+
+@ExtendWith(MockKExtension::class)
+class RunServiceImplTests {
+
+  @Suppress("unused") @MockK private lateinit var containerFactory: RunContainerFactory
+  @MockK private lateinit var workflowService: WorkflowService
+  @MockK private lateinit var runnerApiService: RunnerApiServiceInterface
+  @MockK private lateinit var runRepository: RunRepository
+  @Suppress("unused") @RelaxedMockK private lateinit var csmRbac: CsmRbac
+
+  @InjectMockKs private lateinit var runServiceImpl: RunServiceImpl
+
+  @BeforeTest
+  fun setUp() {
+    MockKAnnotations.init(this)
+  }
+
+  private fun buildRunner(lastRunId: String? = RUN_ID) =
+      Runner(
+          id = RUNNER_ID,
+          name = "runner",
+          createInfo = RunnerEditInfo(timestamp = 0L, userId = "user"),
+          updateInfo = RunnerEditInfo(timestamp = 0L, userId = "user"),
+          solutionId = "sol-id",
+          runTemplateId = "runtemplate-id",
+          organizationId = ORGANIZATION_ID,
+          workspaceId = WORKSPACE_ID,
+          datasets = RunnerDatasets(bases = mutableListOf(), parameter = ""),
+          parametersValues = mutableListOf(),
+          lastRunInfo =
+              LastRunInfo(
+                  lastRunId = lastRunId,
+                  lastRunStatus = LastRunInfo.LastRunStatus.Running,
+              ),
+          validationStatus = RunnerValidationStatus.Validated,
+          security =
+              RunnerSecurity(
+                  default = ROLE_ADMIN,
+                  accessControlList =
+                      mutableListOf(RunnerAccessControl(id = "user", role = ROLE_ADMIN)),
+              ),
+      )
+
+  private fun buildRun(state: RunState?) =
+      Run(
+          id = RUN_ID,
+          organizationId = ORGANIZATION_ID,
+          workspaceId = WORKSPACE_ID,
+          runnerId = RUNNER_ID,
+          state = state,
+          createInfo = RunEditInfo(timestamp = 0L, userId = "user"),
+      )
+
+  @Test
+  fun `onRunStop stops workflow and sets run state to Failed when run is not in terminal state`() {
+    val runner = buildRunner()
+    val run = buildRun(RunState.Running)
+    val runStopRequest = RunStop(this, runner)
+
+    every { runnerApiService.getRunner(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID) } returns runner
+    every { runRepository.findBy(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID, RUN_ID) } returns
+        Optional.of<Run>(run)
+    every { workflowService.getRunStatus(run) } returns RunStatus(id = RUN_ID, phase = "Running")
+    every { workflowService.stopWorkflow(run) } returns RunStatus(id = RUN_ID)
+
+    val savedRunSlot = slot<Run>()
+    every { runRepository.save(capture(savedRunSlot)) } answers { savedRunSlot.captured }
+
+    runServiceImpl.onRunStop(runStopRequest)
+
+    verify(exactly = 1) { workflowService.stopWorkflow(run) }
+    assertEquals(RunState.Failed, savedRunSlot.captured.state)
+  }
+
+  @Test
+  fun `onRunStop throws IllegalStateException when run is Successful (terminal state)`() {
+    val runner = buildRunner()
+    val run = buildRun(RunState.Successful)
+    val runStopRequest = RunStop(this, runner)
+
+    every { runnerApiService.getRunner(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID) } returns runner
+    every { runRepository.findBy(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID, RUN_ID) } returns
+        Optional.of(run)
+
+    assertFailsWith<IllegalStateException> { runServiceImpl.onRunStop(runStopRequest) }
+
+    verify(exactly = 0) { workflowService.stopWorkflow(any()) }
+    verify(exactly = 0) { runRepository.save(any()) }
+  }
+
+  @Test
+  fun `onRunStop throws IllegalStateException when run is Failed (terminal state)`() {
+    val runner = buildRunner()
+    val run = buildRun(RunState.Failed)
+    val runStopRequest = RunStop(this, runner)
+
+    every { runnerApiService.getRunner(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID) } returns runner
+    every { runRepository.findBy(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID, RUN_ID) } returns
+        Optional.of(run)
+
+    assertFailsWith<IllegalStateException> { runServiceImpl.onRunStop(runStopRequest) }
+
+    verify(exactly = 0) { workflowService.stopWorkflow(any()) }
+    verify(exactly = 0) { runRepository.save(any()) }
+  }
+}

runner/src/main/kotlin/com/cosmotech/runner/service/RunnerApiServiceImpl.kt

@@ -122,11 +122,17 @@ internal class RunnerApiServiceImpl(
     val runnerService = getRunnerService().inOrganization(organizationId).inWorkspace(workspaceId)
     val runnerInstance = runnerService.getInstance(runnerId).userHasPermission(PERMISSION_DELETE)
 
+    if (runnerInstance.getRunnerDataObject().status == RunnerStatus.Archived) {
+      throw IllegalStateException(
+          "This runner has already been mark as archived and so is going to be delete"
+      )
+    }
+
+    runnerService.archiveInstance(runnerInstance)
+
     // Set runner status to Archived for future deletion (allow async process)
     runnerInstance.runner.status = RunnerStatus.Archived
     runnerService.saveInstance(runnerInstance.stamp())
-
-    runnerService.archiveInstance(runnerInstance)
   }
 
   override fun listRunners(

runner/src/main/kotlin/com/cosmotech/runner/service/RunnerService.kt

@@ -122,7 +122,6 @@ class RunnerService(
 
   fun archiveInstance(runnerInstance: RunnerInstance) {
     val runner = runnerInstance.getRunnerDataObject()
-
     // Check there are no running runs
     val hasRunningRuns = HasRunningRuns(this, runner.organizationId, runner.workspaceId, runner.id)
     this.eventPublisher.publishEvent(hasRunningRuns)
@@ -131,7 +130,6 @@ class RunnerService(
           "Can't delete runner ${runner.id}: at least one run is still running"
       )
     }
-
     // Update parent and root references to deleted runner
     val newRoots = mutableListOf<Runner>()
     listAllRunnerByParentId(runner.organizationId, runner.workspaceId, runner.id).forEach {
@@ -287,37 +285,44 @@ class RunnerService(
     val runId = startEvent.response ?: throw IllegalStateException("Run Service did not respond")
     runnerInstance.setLastRunInfo(runId)
     runnerRepository.save(runnerInstance.getRunnerDataObject())
+
     return CreatedRun(id = runId)
   }
 
   fun stopLastRunOf(runnerInstance: RunnerInstance) {
     val runner = runnerInstance.getRunnerDataObject()
     runner.lastRunInfo.lastRunId ?: return // No run to stop
-    this.eventPublisher.publishEvent(RunStop(this, runner))
+    val runStopEvent = RunStop(this, runner)
+    this.eventPublisher.publishEvent(runStopEvent)
   }
 
   fun cleanupArchived() {
     // do a cleanup on runner that are archived and have a last run status of successful
     val archivedRunners: List<Runner> = runnerRepository.findAllByStatus(RunnerStatus.Archived)
 
     archivedRunners.forEach { runner ->
-      val cleanupEvent = CleanUpRun(this, runner.lastRunInfo.lastRunId!!)
-      this.eventPublisher.publishEvent(cleanupEvent)
-      if (cleanupEvent.response == true) {
-        runnerRepository.delete(runner)
-        val runnerEvent =
-            RunnerDeleted(
-                this,
-                runner.organizationId,
-                runner.workspaceId,
-                runner.datasets.parameter,
-            )
-        this.eventPublisher.publishEvent(runnerEvent)
+      // If a runner has a last run, we need to cleanup the run before deleting the runner
+      if (runner.lastRunInfo.lastRunId != null) {
+        val cleanupEvent = CleanUpRun(this, runner.lastRunInfo.lastRunId!!)
+        this.eventPublisher.publishEvent(cleanupEvent)
+        if (cleanupEvent.response == false)
+            // The cleanup run process is not finished, we will not delete the runner yet
+            return@forEach
       }
+
+      runnerRepository.delete(runner)
+      val runnerEvent =
+          RunnerDeleted(
+              this,
+              runner.organizationId,
+              runner.workspaceId,
+              runner.datasets.parameter,
+          )
+      this.eventPublisher.publishEvent(runnerEvent)
     }
   }
 
-  @Suppress("TooManyFunctions")
+  @Suppress("TooManyFunctions", "LargeClass")
   inner class RunnerInstance {
     private val roleDefinition: RolesDefinition = getRunnerRolesDefinition()
     lateinit var runner: Runner

runner/src/main/kotlin/com/cosmotech/runner/tasks/RunnerScheduledTasks.kt

@@ -4,10 +4,13 @@ package com.cosmotech.runner.tasks
 
 import com.cosmotech.runner.service.RunnerService
 import java.util.concurrent.TimeUnit
+import org.slf4j.LoggerFactory
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Component
 
+private val logger = LoggerFactory.getLogger(RunnerScheduledTasks::class.java)
+
 @Component
 @ConditionalOnProperty(
     name = ["csm.platform.tasks.cleanUpArchivedRunners.enabled"],
@@ -23,6 +26,7 @@ class RunnerScheduledTasks(
       fixedDelayString = "\${csm.platform.tasks.cleanUpArchivedRunners.delay}",
   )
   fun cleanupArchivedRunners() {
+    logger.debug("Scheduled task: Cleaning up archived runners")
     runnerService.cleanupArchived()
   }
 }