Unique job for visualizations (#3752)

PR adds special kind of jobs for visualizations. It should
- prevent cancelling visualization jobs when the program is re-executed (resulting in the fact that visualization is not showing up)
- omit unnecessary executions of visualization jobs (the case when the user goes through menu items in the component browser)

I skipped the tests because testing of the last scenario involves indeterminism. I.e. when preparing the test, we can't control which of submitted visualization jobs will be actually executed, and which will be cancelled.

engine/runtime-with-instruments/src/test/scala/org/enso/interpreter/test/instrument/RuntimeVisualizationsTest.scala

@@ -1791,9 +1791,9 @@ class RuntimeVisualizationsTest
     context.send(
       Api.Request(requestId, Api.PushContextRequest(contextId, item1))
     )
-    val responses = context.receiveN(n = 5, timeoutSeconds = 60)
+    val responses = context.receiveNIgnoreStdLib(n = 3)
 
-    responses should contain allOf (
+    responses should contain theSameElementsAs Seq(
       Api.Response(requestId, Api.PushContextResponse(contextId)),
       TestMessages.error(
         contextId,
@@ -1803,12 +1803,6 @@ class RuntimeVisualizationsTest
       context.executionComplete(contextId)
     )
 
-    val loadedLibraries = responses.collect {
-      case Api.Response(None, Api.LibraryLoaded(namespace, name, _, _)) =>
-        (namespace, name)
-    }
-    loadedLibraries should contain(("Standard", "Base"))
-
     // attach visualisation
     context.send(
       Api.Request(
@@ -1826,7 +1820,7 @@ class RuntimeVisualizationsTest
         )
       )
     )
-    val attachVisualisationResponses = context.receiveN(5)
+    val attachVisualisationResponses = context.receiveN(4, timeoutSeconds = 60)
     attachVisualisationResponses should contain allOf (
       Api.Response(requestId, Api.VisualisationAttached()),
       context.executionComplete(contextId)

engine/runtime/src/main/scala/org/enso/interpreter/instrument/execution/JobExecutionEngine.scala

@@ -1,7 +1,7 @@
 package org.enso.interpreter.instrument.execution
 
 import org.enso.interpreter.instrument.InterpreterContext
-import org.enso.interpreter.instrument.job.Job
+import org.enso.interpreter.instrument.job.{Job, UniqueJob}
 import org.enso.polyglot.RuntimeServerInfo
 import org.enso.text.Sha3_224VersionCalculator
 
@@ -21,7 +21,7 @@ import scala.util.control.NonFatal
   * @param executionState a state of the runtime
   * @param locking locking capability for runtime
   */
-class JobExecutionEngine(
+final class JobExecutionEngine(
   interpreterContext: InterpreterContext,
   executionState: ExecutionState,
   locking: Locking
@@ -70,8 +70,29 @@ class JobExecutionEngine(
     runInternal(job, backgroundJobExecutor, backgroundJobsRef)
 
   /** @inheritdoc */
-  override def run[A](job: Job[A]): Future[A] =
+  override def run[A](job: Job[A]): Future[A] = {
+    cancelDuplicateJobs(job)
     runInternal(job, jobExecutor, runningJobsRef)
+  }
+
+  private def cancelDuplicateJobs[A](job: Job[A]): Unit = {
+    job match {
+      case job: UniqueJob[_] =>
+        val allJobs =
+          runningJobsRef.updateAndGet(_.filterNot(_.future.isCancelled))
+        allJobs.foreach { runningJob =>
+          runningJob.job match {
+            case jobRef: UniqueJob[_]
+                if jobRef.getClass == job.getClass && jobRef.key == job.key =>
+              runtimeContext.executionService.getLogger
+                .log(Level.FINEST, s"Cancelling duplicate job [$jobRef].")
+              runningJob.future.cancel(jobRef.mayInterruptIfRunning)
+            case _ =>
+          }
+        }
+      case _ =>
+    }
+  }
 
   private def runInternal[A](
     job: Job[A],

engine/runtime/src/main/scala/org/enso/interpreter/instrument/job/Job.scala

@@ -25,5 +25,19 @@ abstract class Job[+A](
   def run(implicit ctx: RuntimeContext): A
 
   override def toString: String = this.getClass.getSimpleName
-
 }
+
+/** The job queue can contain only one job of this type with the same `key`.
+  * When a job of this type is added to the job queue, previous duplicate jobs
+  * are cancelled.
+  *
+  * @param key a unique job key
+  * @param contextIds affected executions contests' ids
+  * @param mayInterruptIfRunning determines if the job may be interruptd when
+  *                              running
+  */
+abstract class UniqueJob[+A](
+  val key: UUID,
+  contextIds: List[UUID],
+  mayInterruptIfRunning: Boolean
+) extends Job[A](contextIds, isCancellable = false, mayInterruptIfRunning)

engine/runtime/src/main/scala/org/enso/interpreter/instrument/job/UpsertVisualisationJob.scala

@@ -37,9 +37,9 @@ class UpsertVisualisationJob(
   visualisationId: VisualisationId,
   expressionId: ExpressionId,
   config: Api.VisualisationConfiguration
-) extends Job[Option[Executable]](
+) extends UniqueJob[Option[Executable]](
+      expressionId,
       List(config.executionContextId),
-      false,
       false
     ) {