feat(jdbc): purge execution queue early

jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java

@@ -939,6 +939,13 @@ private void toExecution(Executor executor, boolean ignoreFailure) {
                 executorService.log(log, false, executor);
             }
 
+            // delete all previous execution messages
+            // we still send the last one so all consumers of terminated executions will be sure to have the latest status
+            // TODO we check 3 times if the exec is terminated, we may want to refactor all this
+            if (executorService.canBePurged(executor)) {
+                ((JdbcQueue<Execution>) executionQueue).deleteByKey(executor.getExecution().getId());
+            }
+
             // emit for other consumer than executor if no failure
             if (hasFailure) {
                 this.executionQueue.emit(executor.getExecution());

jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java

@@ -163,6 +163,18 @@ public void delete(String consumerGroup, T message) throws QueueException {
         // and the queue has its own cleaner, which we better not mess with, as the 'queues' table is selected with a lock.
     }
 
+    public void deleteByKey(String key) throws QueueException {
+        dslContextWrapper.transaction(configuration -> {
+            int deleted = DSL
+                .using(configuration)
+                .delete(this.table)
+                .where(buildTypeCondition(this.cls.getName()))
+                .and(AbstractJdbcRepository.field("key").eq(key))
+                .execute();
+            log.info("Cleaned {} records for key {}", deleted, key);
+        });
+    }
+
     protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, Integer offset) {
         return this.receiveFetch(ctx, consumerGroup, offset, true);
     }