[Fix #1657] Integrate status change events with others

[fjtirado]

The idea is to not continue workflow execution till status change events listeners are completed (note that other listeners are already integrated with workflow)
Remember that listeners errors should not interfere with normal workflow execution, but we do not want the workflow to move on while there are listeners being executed.

Many thanks for submitting your Pull Request ❤️!

What this PR does / why we need it:

Special notes for reviewers:

Additional information (if needed):

[Copilot]

🟡 Changes recommended

There are confirmed correctness issues that can cause hangs and resource leaks (e.g., incomplete futures in WaitExecutor, shared mutable listener state in ListenExecutor, and executor shutdown leakage in DefaultExecutorServiceFactory).

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR aims to ensure workflow execution does not continue until lifecycle-event listeners (notably status-change listeners) have completed, by making lifecycle/status event publication participate in the CompletableFuture execution chain.

Changes:

• Refactors WorkflowMutableInstance.status(...) to return a CompletableFuture and updates multiple execution paths to compose lifecycle events into the main async pipeline.
• Updates task executors (e.g., wait/listen) to delay progression until status-change listener futures complete.
• Adjusts executor-factory defaults and related tests.

File summaries

File	Description
impl/test/src/test/java/io/serverlessworkflow/impl/test/LifeCycleEventsTest.java	Removes custom executor factory usage in the test setup.
impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceInstance.java	Updates persisted-instance start logic to set status/suspension state using new helpers.
impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java	Makes status changes awaitable and composes completion/failure lifecycle events into the workflow future chain.
impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java	Defers default executor-factory initialization until build-time when not explicitly provided.
impl/core/src/main/java/io/serverlessworkflow/impl/executors/WaitExecutor.java	Delays completion until status-change listeners complete (needs exception-safe chaining).
impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java	Attempts to wait for accumulated status-change listener futures (introduces shared mutable state concerns).
impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java	Ensures task failure/cancel events are published before propagating errors downstream.
impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java	Simplifies executor creation (currently introduces a resource-leak via field shadowing).

Review details

Suppressed comments (2)

impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java:188

• This adds the status-listener future to waitingListeners, but because waitingListeners is shared across executor reuse, allOf(...) may wait on stale/unrelated futures and can also throw due to concurrent modification. The listener-future collection should be per internalExecute(...) call and immutable once the completion stage is built.

    waitingListeners.add(
        ((WorkflowMutableInstance) workflow.instance()).status(WorkflowStatus.WAITING));
    EventRegistrationInfo info =
        buildInfo(
            (BiConsumer<CloudEvent, CompletableFuture<WorkflowModel>>)
                ((ce, future) ->
                    processCe(converter.apply(ce), output, workflow, taskContext, future)),
            workflow,
            taskContext);
    workflow.instance().addCancelable(info.completableFuture());
    return info.completableFuture()
        .whenComplete((__, e) -> info.registrations().forEach(eventConsumer::unregister))
        .thenCompose(
            __ ->
                CompletableFuture.allOf(
                    waitingListeners.toArray(new CompletableFuture[waitingListeners.size()])))

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java:80

• status(WorkflowStatus.RUNNING) returns a future that represents completion of status-change listeners, but it is not chained here. As a result, workflow execution (including onWorkflowStarted and subsequent task processing in startExecution) can proceed before status-change listeners finish, defeating the goal of waiting for lifecycle listeners.

  public CompletableFuture<WorkflowModel> start() {
    return startExecution(
        () -> {
          status(WorkflowStatus.RUNNING);
          startedAt = Instant.now();
          return publishEvent(
              workflowContext, l -> l.onWorkflowStarted(new WorkflowStartedEvent(workflowContext)));
        });

• Files reviewed: 8/8 changed files
• Comments generated: 3
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🟡 Changes recommended

There are confirmed ordering/concurrency issues and exception-swallowing paths that prevent reliably waiting for listener completion and can hide listener failures.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java:248

• suspend() now calls status(SUSPENDED) asynchronously and immediately publishes the suspended event without waiting for status-change listeners; this can cause lifecycle events to interleave unexpectedly.

    boolean result = internalSuspend();
    if (result) {
      status(WorkflowStatus.SUSPENDED);
      publishEvent(
          workflowContext, l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext)));
    }

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java:363

• cancel() has the same ordering issue as suspend(): it publishes WorkflowCancelled without waiting for the status-change listeners triggered by status(CANCELLED) to complete.

    boolean result = internalCancel();
    if (result) {
      status(WorkflowStatus.CANCELLED);
      publishEvent(
          workflowContext, l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)));
    }

• Files reviewed: 10/10 changed files
• Comments generated: 7
• Review effort level: Lite

[fjtirado]

This class was a leftover not strictly related with the PR, but since the executors are using the underlying service factory, Im doing the change with the same PR

[Copilot]

🟡 Changes recommended

The updated status/cancel/suspend logic introduces concurrency ordering risks (including a confirmed race in suspendedCheck() and a likely cancellation ordering race) that can lead to incorrect workflow states.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java:409

• internalCancel() cancels registered cancelables before cancel() updates the workflow status to CANCELLED. Because CompletableFuture cancellation can trigger dependent stages synchronously, there is a race where task chains can pass cancel checks before status is updated, causing cancellation to be missed.

    boolean result;
    Collection<CompletableFuture<?>> toCancel = null;
    try {
      statusLock.lock();
      if (TaskExecutorHelper.isActive(status.get())) {
        toCancel = new ArrayList<>(cancelables);
        cancelables.clear();
        result = true;
      } else {
        result = false;
      }
    } finally {
      statusLock.unlock();
    }
    if (result) {
      if (toCancel != null) {
        toCancel.forEach(t -> t.cancel(true));
      }
    }
    return result;

impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java:35

• close() ignores awaitTermination's return value and doesn't restore the interrupt flag; if tasks don’t terminate promptly, threads can remain alive after close (especially visible in tests/embedded runtimes).

  public void close() throws Exception {
    if (!service.isShutdown()) {
      service.shutdown();
      service.awaitTermination(2, TimeUnit.SECONDS);
    }

• Files reviewed: 10/10 changed files
• Comments generated: 3
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

Workflow status updates can still overwrite terminal states (e.g., CANCELLED) due to unconditional getAndSet, risking incorrect cancellation/completion behavior under concurrency.

Review details

Suppressed comments (1)

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java:220

• WorkflowMutableInstance.status(...) unconditionally overwrites the current status via getAndSet(), which allows later calls (e.g., status(RUNNING)/status(WAITING) from executors) to revert terminal states like CANCELLED/COMPLETED/FAULTED. That can make cancellation or completion non-sticky and cause checks like cancelCheck() to miss a cancellation if another thread updates status afterwards. Consider preventing transitions away from terminal states and using compareAndSet to avoid clobbering concurrent updates.

  public CompletableFuture<?> status(WorkflowStatus state) {
    WorkflowStatus prevState = this.status.getAndSet(state);
    return prevState != state
        ? publishEvent(
            workflowContext,

• Files reviewed: 10/10 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

WorkflowMutableInstance.suspendedCheck() currently updates the status to RUNNING in a way that can skip publishing/awaiting the WAITING→RUNNING status-change lifecycle event.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

• Files reviewed: 10/10 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

There are correctness/concurrency issues in WorkflowMutableInstance around RUNNING status-change publishing in suspendedCheck and a cancellation race that can leave newly-added cancelables not cancelled.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (1)

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java:366

• In suspendedCheck(), status is already set to RUNNING under the statusLock via getAndSet(), so the subsequent status(WorkflowStatus.RUNNING) call will see no change and will not publish/await the workflow status-changed listeners. This prevents RUNNING status-change events from being integrated/awaited in this path.

    return prevState != WorkflowStatus.RUNNING
        ? status(WorkflowStatus.RUNNING).thenApply(__ -> t)
        : CompletableFuture.completedFuture(t);

• Files reviewed: 10/10 changed files
• Comments generated: 2
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

WorkflowMutableInstance.startExecution() can still start the workflow twice under concurrent calls due to a non-atomic futureRef.get()/set() sequence, risking duplicate execution.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

• Files reviewed: 10/10 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

There is a confirmed concurrency risk in ListenExecutor due to a non-thread-safe listener-future collection and an executor shutdown implementation that can leak threads on timeout.

Review details

Suppressed comments (2)

impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java:35

• DefaultExecutorServiceFactory.close() may return with executor threads still running if awaitTermination times out, which can leak threads in tests/embedded usage (e.g., InMemoryEvents uses this factory). Escalate to shutdownNow() on timeout and preserve interrupt status when interrupted.

  public void close() throws Exception {
    if (!service.isShutdown()) {
      service.shutdown();
      service.awaitTermination(2, TimeUnit.SECONDS);
    }

impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java:175

• waitingListeners is an ArrayList that is appended to from CloudEvent callbacks; those callbacks can run concurrently (e.g., InMemoryEvents.publish uses CompletableFuture.runAsync on an ExecutorService), so concurrent add/toArray can corrupt the list or throw at runtime. Use a thread-safe collection for waitingListeners.

    Collection<CompletableFuture<?>> waitingListeners = new ArrayList<>();

• Files reviewed: 10/10 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🟢 Approval recommended

The status-change lifecycle events are now consistently awaited along core execution paths (start/wait/listen/end/fault) without letting listener failures affect workflow outcomes, matching the stated intent.

Review details

• Files reviewed: 10/10 changed files
• Comments generated: 0 new
• Review effort level: Lite