fix(triggers): scheduler-owned resubmission of triggers on worker loss

The liveness coordinator re-emitted the persisted WorkerTrigger of a dead worker,
replaying a stale definition and resurrecting disabled or deleted triggers. It now
sends a TriggerWorkerLost event instead; the scheduler releases the lock (guarded
by the holding workerId) and resubmits from the current flow definition. As a
safety net, TriggerReceived for a disabled or deleted trigger re-emits the kill.

Author: fhussonnois

core/src/main/java/io/kestra/core/scheduler/events/TriggerEvent.java

@@ -36,6 +36,7 @@
         @JsonSubTypes.Type(value = ResetTrigger.class, name = "RESET_TRIGGER"),
         @JsonSubTypes.Type(value = SetDisableTrigger.class, name = "SET_DISABLE_TRIGGER"),
         @JsonSubTypes.Type(value = TriggerReceived.class, name = "TRIGGER_RECEIVED"),
+        @JsonSubTypes.Type(value = TriggerWorkerLost.class, name = "TRIGGER_WORKER_LOST"),
         @JsonSubTypes.Type(value = TriggerEvent.Invalid.class, name = "INVALID"),
     }
 )

core/src/main/java/io/kestra/core/scheduler/events/TriggerEventType.java

@@ -16,6 +16,7 @@ public enum TriggerEventType {
     TRIGGER_EVALUATED,
     TRIGGER_EXECUTION_TERMINATED,
     TRIGGER_RECEIVED,
+    TRIGGER_WORKER_LOST,
     // COMMANDS,
     CREATE_BACKFILL_TRIGGER,
     DELETE_BACKFILL_TRIGGER,

core/src/main/java/io/kestra/core/scheduler/events/TriggerWorkerLost.java

@@ -0,0 +1,23 @@
+package io.kestra.core.scheduler.events;
+
+import java.time.Instant;
+
+import io.kestra.core.events.EventId;
+import io.kestra.core.models.triggers.TriggerId;
+
+/**
+ * The worker holding a trigger is gone (crashed or terminated) without reporting a result.
+ * <p>
+ * The scheduler reacts by releasing the trigger's lock so it can be resubmitted from the
+ * current flow definition — or left alone if it is disabled or deleted.
+ */
+public record TriggerWorkerLost(
+    TriggerId id,
+    String workerUid,
+    Instant timestamp,
+    EventId eventId) implements TriggerEvent {
+
+    public TriggerWorkerLost(TriggerId id, String workerUid) {
+        this(id, workerUid, Instant.now(), EventId.create());
+    }
+}

executor/src/main/java/io/kestra/executor/DefaultServiceLivenessCoordinator.java

@@ -15,12 +15,13 @@
 import io.kestra.core.killswitch.KillSwitchService;
 import io.kestra.core.lock.LockService;
 import io.kestra.core.metrics.MetricRegistry;
-import io.kestra.core.models.triggers.RealtimeTriggerInterface;
 import io.kestra.core.models.triggers.TriggerId;
 import io.kestra.core.queues.KeyedDispatchQueueInterface;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.repositories.ServiceInstanceRepositoryInterface;
 import io.kestra.core.runners.*;
+import io.kestra.core.scheduler.events.TriggerWorkerLost;
+import io.kestra.core.scheduler.queue.TriggerEventQueue;
 import io.kestra.core.scheduler.vnodes.VNodeController;
 import io.kestra.core.server.*;
 import io.kestra.core.utils.IdUtils;
@@ -66,6 +67,7 @@ public class DefaultServiceLivenessCoordinator extends AbstractServiceLivenessTa
     private final KillSwitchService killSwitchService;
     private final KeyedDispatchQueueInterface<WorkerJobEvent> workerJobEventQueue;
     private final WorkerJobRunningStateStore workerJobRunningStateStore;
+    private final TriggerEventQueue triggerEventQueue;
     private final MetricRegistry metricRegistry;
     private final VNodeController vNodeController;
     private Counter workerJobResubmitCounter;
@@ -92,6 +94,7 @@ public DefaultServiceLivenessCoordinator(final ServiceLivenessStore store,
         final KillSwitchService killSwitchService,
         final KeyedDispatchQueueInterface<WorkerJobEvent> workerJobEventQueue,
         final WorkerJobRunningStateStore workerJobRunningStateStore,
+        final TriggerEventQueue triggerEventQueue,
         final ServerConfig serverConfig,
         final MetricRegistry metricRegistry,
         final VNodeController vNodeController) {
@@ -103,6 +106,7 @@ public DefaultServiceLivenessCoordinator(final ServiceLivenessStore store,
         this.killSwitchService = killSwitchService;
         this.workerJobEventQueue = workerJobEventQueue;
         this.workerJobRunningStateStore = workerJobRunningStateStore;
+        this.triggerEventQueue = triggerEventQueue;
         this.lockService = lockService;
         this.metricRegistry = metricRegistry;
         this.purgeRetention = serverConfig.service() != null && serverConfig.service().purge() != null
@@ -176,10 +180,10 @@ protected void handleAllWorkersForUncleanShutdown(Instant now) {
                     reEmitWorkerJobsForWorker(txContext, serviceInstance.uid());
                 }
             } else if (serviceInstance.is(Service.ServiceState.TERMINATED_GRACEFULLY)) {
-                // realtime triggers need to be resubmitted even when terminated gracefully
+                // triggers whose terminal result was not flushed need to be released even when terminated gracefully
                 if (serviceInstance.config().workerTaskRestartStrategy().isRestartable()) {
-                    log.info("Trigger realtime trigger restart for terminated gracefully worker: {}.", serviceInstance.uid());
-                    reEmitRealtimeTriggerForWorker(txContext, serviceInstance.uid());
+                    log.info("Trigger restart for terminated gracefully worker: {}.", serviceInstance.uid());
+                    notifyTriggersLostForWorker(txContext, serviceInstance.uid());
                 }
             }
 
@@ -257,10 +261,12 @@ private void reEmitWorkerJobsForWorker(final TransactionContext txContext, final
         });
     }
 
-    private void reEmitRealtimeTriggerForWorker(final TransactionContext txContext, final String id) {
+    private void notifyTriggersLostForWorker(final TransactionContext txContext, final String id) {
         workerJobRunningStateStore.processWorkerJobsForDeadWorker(txContext, id, (txContext2, workerJobRunning) ->
         {
-            resubmitRealtimeTrigger(txContext2, workerJobRunning);
+            if (workerJobRunning instanceof WorkerTriggerRunning workerTriggerRunning) {
+                notifyTriggerWorkerLost(txContext2, workerTriggerRunning);
+            }
         });
     }
 
@@ -393,14 +399,7 @@ private void resubmitWorkerJobRunning(TransactionContext txContext, WorkerJobRun
 
         // WorkerTriggerRunning
         if (workerJobRunning instanceof WorkerTriggerRunning workerTriggerRunning) {
-            resubmitWorkerTrigger(workerTriggerRunning);
-        }
-    }
-
-    private void resubmitRealtimeTrigger(TransactionContext txContext, WorkerJobRunning workerJobRunning) {
-        // we only resubmit realtime triggers
-        if (workerJobRunning instanceof WorkerTriggerRunning workerTriggerRunning && workerTriggerRunning.getTrigger() instanceof RealtimeTriggerInterface) {
-            resubmitWorkerTrigger(workerTriggerRunning);
+            notifyTriggerWorkerLost(txContext, workerTriggerRunning);
         }
     }
 
@@ -435,30 +434,25 @@ private void resubmitWorkerTask(TransactionContext txContext, WorkerTaskRunning
         }
     }
 
-    private void resubmitWorkerTrigger(WorkerTriggerRunning workerTriggerRunning) {
-        try {
-            String raw = workerTriggerRunning.getWorkerInstance().workerQueueId();
-            String workerQueueId = (raw == null || raw.isEmpty()) ? null : raw;
-            WorkerTrigger workerTrigger = WorkerTrigger.builder()
-                .trigger(workerTriggerRunning.getTrigger())
-                .data(workerTriggerRunning.getData())
-                .build();
-            workerJobEventQueue.emit(workerQueueId, WorkerJobEvent.of(workerTrigger, workerQueueId));
-            Logs.logTrigger(
-                workerTrigger.triggerId(),
-                Level.WARN,
-                "Re-emitting WorkerTrigger."
-            );
-        } catch (QueueException e) {
-            Logs.logTrigger(
-                TriggerId.of(
-                    workerTriggerRunning.getData().tenantId(), workerTriggerRunning.getData().namespace(), workerTriggerRunning.getData().flowId(),
-                    workerTriggerRunning.getTrigger().getId()
-                ),
-                Level.ERROR,
-                "Unable to re-emit WorkerTrigger.",
-                e
-            );
-        }
+    /**
+     * The worker holding the trigger is gone: delete its running entry and notify the scheduler,
+     * which owns resubmission. Re-emitting the persisted job from here would replay a stale
+     * trigger definition and ignore a disabled or deleted state.
+     */
+    private void notifyTriggerWorkerLost(TransactionContext txContext, WorkerTriggerRunning workerTriggerRunning) {
+        TriggerId triggerId = TriggerId.of(
+            workerTriggerRunning.getData().tenantId(),
+            workerTriggerRunning.getData().namespace(),
+            workerTriggerRunning.getData().flowId(),
+            workerTriggerRunning.getTrigger().getId()
+        );
+        workerJobRunningStateStore.deleteByKey(txContext, workerTriggerRunning.uid());
+        triggerEventQueue.send(new TriggerWorkerLost(triggerId, workerTriggerRunning.getWorkerInstance().uid()));
+        Logs.logTrigger(
+            triggerId,
+            Level.WARN,
+            "Worker '{}' lost, notifying the scheduler.",
+            workerTriggerRunning.getWorkerInstance().uid()
+        );
     }
 }

executor/src/test/java/io/kestra/executor/AbstractServiceLivenessCoordinatorTest.java

@@ -41,9 +41,9 @@
 import io.kestra.core.runners.WorkerTrigger;
 import io.kestra.core.runners.WorkerTriggerData;
 import io.kestra.core.worker.models.WorkerTriggerResult;
-import io.kestra.core.scheduler.events.TriggerEvaluated;
 import io.kestra.core.scheduler.events.TriggerEvent;
 import io.kestra.core.scheduler.events.TriggerReceived;
+import io.kestra.core.scheduler.events.TriggerWorkerLost;
 import io.kestra.core.scheduler.model.TriggerState;
 import io.kestra.core.server.ServerConfig;
 import io.kestra.core.server.ServiceStateChangeEvent;
@@ -218,15 +218,15 @@ void shouldNotResubmitTaskForIgnoredExecution() throws Exception {
 
     @ParameterizedTest
     @ValueSource(strings = { WORKER_QUEUE_UID, "<null>" })
-    public void shouldResubmitTriggerWhenWorkerIsStopped(String workerQueueId) throws Exception {
+    public void shouldNotifyTriggerWorkerLostWhenWorkerIsStopped(String workerQueueId) throws Exception {
         workerQueueId = "<null>".equals(workerQueueId) ? null : workerQueueId;
         // Given - create first worker.
         WorkerAgent worker = (WorkerAgent) newWorker();
         worker.start(1);
 
         WorkerTrigger workerTrigger = workerTrigger(Duration.ofSeconds(5), workerQueueId);
 
-        CountDownLatch evaluatedLatch = new CountDownLatch(1);
+        CountDownLatch lostLatch = new CountDownLatch(1);
         CountDownLatch receivedLatch = new CountDownLatch(1);
         triggerEventQueue.addListener(event ->
         {
@@ -236,8 +236,8 @@ public void shouldResubmitTriggerWhenWorkerIsStopped(String workerQueueId) throw
             if (event instanceof TriggerReceived) {
                 receivedLatch.countDown();
             }
-            if (event instanceof TriggerEvaluated) {
-                evaluatedLatch.countDown();
+            if (event instanceof TriggerWorkerLost) {
+                lostLatch.countDown();
             }
         });
 
@@ -250,8 +250,8 @@ public void shouldResubmitTriggerWhenWorkerIsStopped(String workerQueueId) throw
         WorkerAgent newWorker = (WorkerAgent) newWorker();
         newWorker.start(1);
 
-        // THEN
-        assertThat(evaluatedLatch.await(30, TimeUnit.SECONDS)).isTrue();
+        // THEN - the scheduler is notified instead of the job being re-emitted to a worker.
+        assertThat(lostLatch.await(30, TimeUnit.SECONDS)).isTrue();
         newWorker.close();
     }
 

scheduler/src/main/java/io/kestra/scheduler/TriggerEventHandler.java

@@ -25,6 +25,7 @@
 import io.kestra.core.models.triggers.Backfill;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.models.triggers.TriggerContext;
+import io.kestra.core.models.triggers.TriggerId;
 import io.kestra.core.queues.BroadcastQueueInterface;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.runners.RunContext;
@@ -42,6 +43,7 @@
 import io.kestra.core.scheduler.events.TriggerFlowRevisionUpdated;
 import io.kestra.core.scheduler.events.TriggerReceived;
 import io.kestra.core.scheduler.events.TriggerUpdated;
+import io.kestra.core.scheduler.events.TriggerWorkerLost;
 import io.kestra.core.scheduler.model.TriggerState;
 import io.kestra.core.scheduler.model.TriggerType;
 import io.kestra.core.scheduler.service.TriggerExecutionPublisher;
@@ -121,6 +123,7 @@ private void doHandle(Clock clock, Integer vNode, TriggerEvent event) {
             case TriggerExecutionTerminated evt -> onTriggerExecutionTerminated(clock, evt);
             case TriggerEvaluated evt -> onTriggerEvaluated(clock, evt);
             case TriggerReceived evt -> onTriggerReceived(clock, evt);
+            case TriggerWorkerLost evt -> onTriggerWorkerLost(clock, evt);
             // Commands
             case CreateBackfillTrigger evt -> onCreateBackfill(clock, evt);
             case SetPauseBackfillTrigger evt -> onSetPauseBackfillTrigger(clock, evt);
@@ -317,17 +320,55 @@ void onTriggerEvaluated(Clock clock, TriggerEvaluated event) {
     }
 
     /**
-     * Handler method for {@link ResetTrigger}.
+     * Handler method for {@link TriggerReceived}.
      *
      * @param event the event.
      */
     void onTriggerReceived(Clock clock, TriggerReceived event) {
+        Optional<TriggerState> maybeState = findTriggerState(event);
+        if (maybeState.isEmpty()) {
+            // The trigger was deleted while its worker job was in flight: kill the instance
+            // the worker just started. Only do so when the state is truly missing, not when
+            // the event was de-duplicated.
+            if (triggerStateStore.findById(event.id()).isEmpty()) {
+                sendExecutionKilled(event.id());
+            }
+            return;
+        }
+        TriggerState state = maybeState.get();
+        // The trigger was disabled while its worker job was in flight: the kill broadcast
+        // found no holder at that time, so kill the instance now that a worker reports it.
+        if (state.isDisabled()) {
+            maySendExecutionKilled(state);
+        }
+        triggerStateStore.save(
+            state
+                .lastEventId(clock, event.eventId())
+                .workerId(clock, event.workerId())
+        );
+    }
+
+    /**
+     * Handler method for {@link TriggerWorkerLost}.
+     * <p>
+     * The worker holding the trigger is gone without reporting a result: release the lock so an
+     * eligible trigger is resubmitted from the current flow definition, while a disabled one stays off.
+     *
+     * @param event the event.
+     */
+    void onTriggerWorkerLost(Clock clock, TriggerWorkerLost event) {
         findTriggerState(event).ifPresent(state ->
         {
-            state = state
-                .lastEventId(clock, event.eventId())
-                .workerId(clock, event.workerId());
-            triggerStateStore.save(state);
+            if (state.getWorkerId() != null && !state.getWorkerId().equals(event.workerUid())) {
+                // The trigger is already held by another worker.
+                return;
+            }
+            triggerStateStore.save(
+                state
+                    .lastEventId(clock, event.eventId())
+                    .locked(clock, false)
+                    .workerId(clock, null)
+            );
         });
     }
 
@@ -404,22 +445,26 @@ void onTriggerDeleted(TriggerDeleted event) {
      */
     private void maySendExecutionKilled(TriggerState state) {
         if (TriggerType.REALTIME.equals(state.getType())) {
-            try {
-                this.executionKilledQueue.emit(
-                    ExecutionKilledTrigger
-                        .builder()
-                        // Trigger kills are not processed by the Executor: emit them directly in the
-                        // EXECUTED state, the only state forwarded to the workers.
-                        .state(ExecutionKilled.State.EXECUTED)
-                        .tenantId(state.getTenantId())
-                        .namespace(state.getNamespace())
-                        .flowId(state.getFlowId())
-                        .triggerId(state.getTriggerId())
-                        .build()
-                );
-            } catch (QueueException e) {
-                Logs.logTrigger(state, Level.WARN, "Cannot kill a real-time trigger, it will continue processing until Kestra is restarted. Cause: {}", e.getMessage(), e);
-            }
+            sendExecutionKilled(state);
+        }
+    }
+
+    private void sendExecutionKilled(TriggerId id) {
+        try {
+            this.executionKilledQueue.emit(
+                ExecutionKilledTrigger
+                    .builder()
+                    // Trigger kills are not processed by the Executor: emit them directly in the
+                    // EXECUTED state, the only state forwarded to the workers.
+                    .state(ExecutionKilled.State.EXECUTED)
+                    .tenantId(id.getTenantId())
+                    .namespace(id.getNamespace())
+                    .flowId(id.getFlowId())
+                    .triggerId(id.getTriggerId())
+                    .build()
+            );
+        } catch (QueueException e) {
+            Logs.logTrigger(id, Level.WARN, "Cannot kill a real-time trigger, it will continue processing until Kestra is restarted. Cause: {}", e.getMessage(), e);
         }
     }