fix(ui,webserver): send a start execution to make sure the SSE connec…

…tion is initialized
kestra-io · Jan 22, 2025 · 8244a16 · 8244a16
1 parent e5ae20d
commit 8244a16
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 27 deletions.
diff --git a/ui/src/components/executions/ExecutionRoot.vue b/ui/src/components/executions/ExecutionRoot.vue
@@ -101,28 +101,31 @@
                                 self.closeSSE();
                             }
 
-                            let execution = JSON.parse(event.data);
+                            // we are receiving a first "fake" event to force initializing the connection: ignoring it
+                            if (event.lastEventId !== "start") {
+                                let execution = JSON.parse(event.data);
 
-                            if (!this.flow ||
-                                execution.flowId !== this.flow.id ||
-                                execution.namespace !== this.flow.namespace ||
-                                execution.flowRevision !== this.flow.revision
-                            ) {
-                                this.$store.dispatch(
-                                    "flow/loadFlow",
-                                    {
+                                if (!this.flow ||
+                                    execution.flowId !== this.flow.id ||
+                                    execution.namespace !== this.flow.namespace ||
+                                    execution.flowRevision !== this.flow.revision
+                                ) {
+                                    this.$store.dispatch(
+                                        "flow/loadFlow",
+                                        {
+                                            namespace: execution.namespace,
+                                            id: execution.flowId,
+                                            revision: execution.flowRevision
+                                        }
+                                    );
+                                    this.$store.dispatch("flow/loadRevisions", {
                                         namespace: execution.namespace,
-                                        id: execution.flowId,
-                                        revision: execution.flowRevision
-                                    }
-                                );
-                                this.$store.dispatch("flow/loadRevisions", {
-                                    namespace: execution.namespace,
-                                    id: execution.flowId
-                                })
-                            }
+                                        id: execution.flowId
+                                    })
+                                }
 
-                            this.$store.commit("execution/setExecution", execution);
+                                this.$store.commit("execution/setExecution", execution);
+                            }
                         }
                         // sse.onerror doesnt return the details of the error
                         // but as our emitter can only throw an error on 404

diff --git a/ui/src/components/executions/Topology.vue b/ui/src/components/executions/Topology.vue
@@ -210,18 +210,28 @@
                         this.sseBySubflow[subflow] = sse;
                         sse.onmessage = (event) => {
                             if (event && event.lastEventId === "end") {
-                                sse.close();
+                                this.closeSubExecutionSSE(subflow);
                             }
 
-                            const previousExecution = this.subflowsExecutions[subflow];
-                            this.$store.commit("execution/addSubflowExecution", {subflow, execution: JSON.parse(event.data)});
+                            // we are receiving a first "fake" event to force initializing the connection: ignoring it
+                            if (event.lastEventId !== "start") {
+                                const previousExecution = this.subflowsExecutions[subflow];
+                                this.$store.commit("execution/addSubflowExecution", {subflow, execution: JSON.parse(event.data)});
 
-                            // add subflow execution id to graph
-                            if(previousExecution === undefined) {
-                                this.loadGraph(true);
+                                // add subflow execution id to graph
+                                if(previousExecution === undefined) {
+                                    this.loadGraph(true);
+                                }
                             }
                         };
                     });
+            },
+            closeSubExecutionSSE(subflow) {
+                const sse = this.sseBySubflow[subflow];
+                if (sse) {
+                    sse.close();
+                    delete this.sseBySubflow[subflow];
+                }
             }
         }
     };

diff --git a/ui/src/components/logs/TaskRunDetails.vue b/ui/src/components/logs/TaskRunDetails.vue
@@ -327,7 +327,14 @@
                     .then(sse => {
                         this.executionSSE = sse;
                         this.executionSSE.onmessage = async (event) => {
-                            this.followedExecution = JSON.parse(event.data);
+                            if (event && event.lastEventId === "end") {
+                                sse.close();
+                            }
+
+                            // we are receiving a first "fake" event to force initializing the connection: ignoring it
+                            if (event.lastEventId !== "start") {
+                                this.followedExecution = JSON.parse(event.data);
+                            }
                         }
                     });
             },

diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
@@ -1114,8 +1114,11 @@ public Flux<Event<Execution>> follow(
 
         return Flux
             .<Event<Execution>>create(emitter -> {
+                // send a first "empty" event so the SSE is correctly initialized in the frontend in case there are no logs
+                emitter.next(Event.of(Execution.builder().id(executionId).build()).id("start"));
+
                 // already finished execution
-                Execution execution = null;
+                Execution execution;
                 try {
                     execution = Await.until(
                         () -> executionRepository.findById(tenantService.resolveTenant(), executionId).orElse(null),