Engine should send notification about node status

app/gui/controller/engine-protocol/src/language_server/types.rs

@@ -236,6 +236,11 @@ pub enum ExpressionUpdatePayload {
         message: String,
         trace:   Vec<ExpressionId>,
     },
+    #[serde(rename_all = "camelCase")]
+    Pending {
+        message: Option<String>,
+        progress:Option<f64>,
+    },
 }
 
 

app/gui/src/presenter/graph/state.rs

@@ -299,7 +299,7 @@ impl Expressions {
 ///
 /// This structure keeps the information how the particular graph elements received from controllers
 /// are represented in the view. It also handles updates from the controllers and
-/// the view in `update_from_controller` and `update_from_view` respectively.  
+/// the view in `update_from_controller` and `update_from_view` respectively.
 #[derive(Clone, Debug, Default)]
 pub struct State {
     nodes:       RefCell<Nodes>,
@@ -492,6 +492,7 @@ impl<'a> ControllerChange<'a> {
             None | Some(Value) => None,
             Some(DataflowError { trace }) => Some((Kind::Dataflow, None, trace)),
             Some(Panic { message, trace }) => Some((Kind::Panic, Some(message), trace)),
+            Some(Pending { message: _, progress: _ }) => Some((Kind::Panic, Some("Pending...".to_string()), Vec::new())),
         }?;
         let propagated = if kind == Kind::Panic {
             let nodes = self.nodes.borrow();
@@ -678,15 +679,15 @@ impl<'a> ViewChange<'a> {
 
 impl<'a> ViewChange<'a> {
     /// If the connections does not already exist, it is created and corresponding to-be-created
-    /// Ast connection is returned.  
+    /// Ast connection is returned.
     pub fn create_connection(&self, connection: view::graph_editor::Edge) -> Option<AstConnection> {
         let source = connection.source()?;
         let target = connection.target()?;
         self.create_connection_from_endpoints(connection.id(), source, target)
     }
 
     /// If the connections with provided endpoints does not already exist, it is created and
-    /// corresponding to-be-created Ast connection is returned.  
+    /// corresponding to-be-created Ast connection is returned.
     pub fn create_connection_from_endpoints(
         &self,
         connection: ViewConnection,

engine/language-server/src/main/scala/org/enso/languageserver/runtime/ContextEventsListener.scala

@@ -247,6 +247,9 @@ final class ContextEventsListener(
       case Api.ExpressionUpdate.Payload.Value() =>
         ContextRegistryProtocol.ExpressionUpdate.Payload.Value
 
+      case Api.ExpressionUpdate.Payload.Pending(m, p) =>
+        ContextRegistryProtocol.ExpressionUpdate.Payload.Pending(m, p)
+
       case Api.ExpressionUpdate.Payload.DataflowError(trace) =>
         ContextRegistryProtocol.ExpressionUpdate.Payload.DataflowError(trace)
 

engine/language-server/src/main/scala/org/enso/languageserver/runtime/ContextRegistryProtocol.scala

@@ -161,6 +161,8 @@ object ContextRegistryProtocol {
       /** An information about computed expression. */
       case object Value extends Payload
 
+      case class Pending(message: Option[String], progress: Option[Double]) extends Payload;
+
       /** Indicates that the expression was computed to an error.
         *
         * @param trace the list of expressions leading to the root error.
@@ -186,6 +188,8 @@ object ContextRegistryProtocol {
 
         val Value = "Value"
 
+        val Pending = "Pending"
+
         val DataflowError = "DataflowError"
 
         val Panic = "Panic"
@@ -210,6 +214,13 @@ object ContextRegistryProtocol {
               .deepMerge(
                 Json.obj(CodecField.Type -> PayloadType.Panic.asJson)
               )
+
+          case m: Payload.Pending =>
+            Encoder[Payload.Pending]
+            .apply(m)
+              .deepMerge(
+                Json.obj(CodecField.Type -> PayloadType.Pending.asJson)
+              )
         }
 
       implicit val decoder: Decoder[Payload] =
@@ -223,6 +234,9 @@ object ContextRegistryProtocol {
 
             case PayloadType.Panic =>
               Decoder[Payload.Panic].tryDecode(cursor)
+
+            case PayloadType.Pending =>
+              Decoder[Payload.Pending].tryDecode(cursor)
           }
         }
     }

engine/polyglot-api/src/main/scala/org/enso/polyglot/runtime/Runtime.scala

@@ -367,6 +367,10 @@ object Runtime {
           new JsonSubTypes.Type(
             value = classOf[Payload.Panic],
             name  = "expressionUpdatePayloadPanic"
+          ),
+          new JsonSubTypes.Type(
+            value = classOf[Payload.Pending],
+            name  = "expressionUpdatePayloadPending"
           )
         )
       )
@@ -378,6 +382,11 @@ object Runtime {
           */
         case class Value() extends Payload
 
+        /** TBD
+          */
+        case class Pending(message: Option[String], progress: Option[Double])
+            extends Payload;
+
         /** Indicates that the expression was computed to an error.
           *
           * @param trace the list of expressions leading to the root error.

engine/runtime-instrument-id-execution/src/main/java/org/enso/interpreter/instrument/IdExecutionInstrument.java

@@ -186,12 +186,11 @@ public void onReturnValue(EventContext context, VirtualFrame frame, Object resul
         FunctionCallInfo cachedCall = cache.getCall(nodeId);
         ProfilingInfo[] profilingInfo = new ProfilingInfo[] {new ExecutionTime(nanoTimeElapsed)};
 
+        System.out.println("onReturnValue " + nodeId + " value: " + result + " type: " + resultType);
         ExpressionValue expressionValue =
             new ExpressionValue(
                 nodeId, result, resultType, cachedType, call, cachedCall, profilingInfo, false);
-        if (expressionValue.isTypeChanged() || expressionValue.isFunctionCallChanged()) {
-          syncState.setExpressionUnsync(nodeId);
-        }
+        syncState.setExpressionUnsync(nodeId);
         syncState.setVisualisationUnsync(nodeId);
 
         // Panics are not cached because a panic can be fixed by changing seemingly unrelated code,

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

@@ -729,7 +729,9 @@ class RuntimeErrorsTest
         )
       )
     )
-    context.receiveN(1) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      1
+    ) should contain theSameElementsAs Seq(
       context.executionComplete(contextId)
     )
     context.consumeOut shouldEqual List("(Error: MyError2)")
@@ -921,7 +923,9 @@ class RuntimeErrorsTest
       )
     )
 
-    context.receiveNIgnoreStdLib(4) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      4
+    ) should contain theSameElementsAs Seq(
       TestMessages.update(contextId, xId, ConstantsGen.INTEGER),
       TestMessages.update(contextId, yId, ConstantsGen.INTEGER),
       TestMessages.update(contextId, mainResId, ConstantsGen.NOTHING),
@@ -1263,7 +1267,9 @@ class RuntimeErrorsTest
         )
       )
     )
-    context.receiveN(4) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      4
+    ) should contain theSameElementsAs Seq(
       TestMessages.update(
         contextId,
         xId,
@@ -1364,7 +1370,9 @@ class RuntimeErrorsTest
         )
       )
     )
-    context.receiveN(3) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      3
+    ) should contain theSameElementsAs Seq(
       TestMessages.update(
         contextId,
         xId,

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

@@ -1174,7 +1174,9 @@ class RuntimeServerTest
         )
       )
     )
-    context.receiveN(1) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      1
+    ) should contain theSameElementsAs Seq(
       context.executionComplete(contextId)
     )
     context.consumeOut shouldEqual List("6")

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

@@ -1148,7 +1148,9 @@ class RuntimeSuggestionUpdatesTest
         )
       )
     )
-    context.receiveN(2) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      2
+    ) should contain theSameElementsAs Seq(
       Api.Response(
         Api.SuggestionsDatabaseModuleUpdateNotification(
           module = moduleName,

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

@@ -567,6 +567,7 @@ class RuntimeVisualizationsTest
       Api.Request(requestId, Api.PushContextRequest(contextId, item1))
     )
 
+    System.out.println("Begin check")
     context.receiveNIgnoreStdLib(5) should contain theSameElementsAs Seq(
       Api.Response(requestId, Api.PushContextResponse(contextId)),
       context.Main.Update.mainX(contextId),
@@ -629,7 +630,7 @@ class RuntimeVisualizationsTest
       )
     )
 
-    val editFileResponse = context.receiveN(2)
+    val editFileResponse = context.receiveNIgnoreExpressionUpdates(2)
     editFileResponse should contain(
       context.executionComplete(contextId)
     )
@@ -754,7 +755,7 @@ class RuntimeVisualizationsTest
       )
     )
 
-    val editFileResponse = context.receiveN(2)
+    val editFileResponse = context.receiveNIgnoreExpressionUpdates(2)
     editFileResponse should contain(
       context.executionComplete(contextId)
     )
@@ -1139,7 +1140,9 @@ class RuntimeVisualizationsTest
       )
     )
 
-    context.receiveN(1) should contain theSameElementsAs Seq(
+    context.receiveNIgnoreExpressionUpdates(
+      1
+    ) should contain theSameElementsAs Seq(
       context.executionComplete(contextId)
     )
   }

engine/runtime/src/main/scala/org/enso/interpreter/instrument/CacheInvalidation.scala

@@ -114,9 +114,10 @@ object CacheInvalidation {
     */
   def runAll(
     stack: Iterable[InstrumentFrame],
-    instructions: Iterable[CacheInvalidation]
+    instructions: Iterable[CacheInvalidation],
+    invalidatedKeys: java.util.Set[UUID]
   ): Unit =
-    instructions.foreach(run(stack, _))
+    instructions.foreach(run(stack, _, invalidatedKeys))
 
   /** Run a sequence of invalidation instructions on all visualisations.
     *
@@ -157,13 +158,14 @@ object CacheInvalidation {
     */
   def run(
     stack: Iterable[InstrumentFrame],
-    instruction: CacheInvalidation
+    instruction: CacheInvalidation,
+    invalidatedKeys: java.util.Set[UUID]
   ): Unit = {
     val frames = instruction.elements match {
       case StackSelector.All => stack
       case StackSelector.Top => stack.headOption.toSeq
     }
-    run(frames, instruction.command, instruction.indexes)
+    run(frames, instruction.command, instruction.indexes, invalidatedKeys)
   }
 
   /** Run cache invalidation of a multiple instrument frames.
@@ -175,9 +177,12 @@ object CacheInvalidation {
   private def run(
     frames: Iterable[InstrumentFrame],
     command: Command,
-    indexes: Set[IndexSelector]
+    indexes: Set[IndexSelector],
+    invalidatedKeys: java.util.Set[UUID]
   ): Unit = {
-    frames.foreach(frame => run(frame.cache, frame.syncState, command, indexes))
+    frames.foreach(frame =>
+      run(frame.cache, frame.syncState, command, indexes, invalidatedKeys)
+    )
   }
 
   /** Run cache invalidation of a single instrument frame.
@@ -221,21 +226,25 @@ object CacheInvalidation {
     cache: RuntimeCache,
     syncState: UpdatesSynchronizationState,
     command: Command,
-    indexes: Set[IndexSelector]
+    indexes: Set[IndexSelector],
+    invalidatedKeys: java.util.Set[UUID]
   ): Unit =
     command match {
       case Command.InvalidateAll =>
+        invalidatedKeys.addAll(cache.getKeys)
         cache.clear()
         indexes.foreach(clearIndex(_, cache))
       case Command.InvalidateKeys(keys) =>
         keys.foreach { key =>
           cache.remove(key)
+          invalidatedKeys.add(key)
           indexes.foreach(clearIndexKey(key, _, cache))
         }
       case Command.InvalidateStale(scope) =>
         val staleKeys = cache.getKeys.asScala.diff(scope.toSet)
         staleKeys.foreach { key =>
           cache.remove(key)
+          invalidatedKeys.add(key)
           indexes.foreach(clearIndexKey(key, _, cache))
           syncState.invalidate(key)
         }

engine/runtime/src/main/scala/org/enso/interpreter/instrument/command/RecomputeContextCmd.scala

@@ -43,7 +43,11 @@ class RecomputeContextCmd(
         val cacheInvalidationCommands = request.expressions.toSeq
           .map(CacheInvalidation.Command(_))
           .map(CacheInvalidation(CacheInvalidation.StackSelector.Top, _))
-        CacheInvalidation.runAll(stack, cacheInvalidationCommands)
+        CacheInvalidation.runAll(
+          stack,
+          cacheInvalidationCommands,
+          new java.util.HashSet[java.util.UUID]()
+        )
         reply(Api.RecomputeContextResponse(request.contextId))
         true
       }