From 31be3f32c619e08bcc46aeaeb40d7493e94ecd82 Mon Sep 17 00:00:00 2001 From: konlue <2362586639@qq.com> Date: Sun, 15 Nov 2026 11:33:20 +0800 Subject: [PATCH] fix(scheduling): make ScheduledAgentTask start/stop atomic state transitions (issue #215) start/stop composed their lifecycle from independent volatile started/stopped flags, so the transitions were not atomic: two concurrent starts could both pass the guard and publish two schedules, a stop racing with the schedule publication could miss the future published afterwards, a stop followed by start resurrected the stopped task, and a failed start left the registration behind. The lifecycle is now a single CAS-able state (NEW, STARTING, RUNNING, STOPPING, STOPPED, FAILED). start() claims NEW->STARTING up front, so a second start is rejected while the first is still publishing; a stop winning the STARTING window makes the late future cancelled instead of live; a scheduling failure releases the registration and lands in FAILED. stop() is idempotent through a single winner transition. The derived isStarted/isStopped keep their previous observable semantics, and all existing lifecycle/ownership tests pass unchanged. Deterministic interleaving tests gate the scheduler publication with latches: double start schedules once, start after stop is rejected, failed start unregisters, stop during start cancels the late future. --- .../graph/scheduling/ScheduledAgentTask.java | 140 +++++-- ...ScheduledAgentTaskAtomicStartStopTest.java | 372 ++++++++++++++++++ 2 files changed, 480 insertions(+), 32 deletions(-) create mode 100644 argi-orchestration/argi-workflow-graph-core/src/test/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTaskAtomicStartStopTest.java diff --git a/argi-orchestration/argi-workflow-graph-core/src/main/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTask.java b/argi-orchestration/argi-workflow-graph-core/src/main/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTask.java index 728891dc6..61d46a4ac 100644 --- a/argi-orchestration/argi-workflow-graph-core/src/main/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTask.java +++ b/argi-orchestration/argi-workflow-graph-core/src/main/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTask.java @@ -21,6 +21,7 @@ import java.util.Optional; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import io.github.agentic.ai.workflow.graph.CompiledGraph; import io.github.agentic.ai.workflow.graph.OverAllState; @@ -51,9 +52,35 @@ public class ScheduledAgentTask { private volatile ScheduledFuture scheduledFuture; - private volatile boolean started = false; + /** + * Atomic lifecycle state (issue #215): NEW → STARTING → RUNNING → STOPPING → + * STOPPED, with FAILED as the start-failure terminal. A single CAS-able state + * replaces the previous independent started/stopped flags, so concurrent start + * and stop transitions cannot interleave. + */ + enum ScheduleState { + + /** Created and registered, not yet scheduled. */ + NEW, + + /** A start is publishing the schedule; stop may still win the race. */ + STARTING, + + /** The schedule is live. */ + RUNNING, + + /** stop() is cancelling and unregistering. */ + STOPPING, + + /** Terminal after a successful stop; start is rejected. */ + STOPPED, - private volatile boolean stopped = false; + /** Terminal after a failed start; registration has been released. */ + FAILED + + } + + private final AtomicReference state = new AtomicReference<>(ScheduleState.NEW); private final AtomicBoolean unregistered = new AtomicBoolean(false); @@ -73,35 +100,51 @@ public ScheduledAgentTask(CompiledGraph graph, ScheduleConfig config) { * Start the scheduled execution */ public ScheduledAgentTask start() { - if (started) { - throw new IllegalStateException("Schedule already started"); + if (!state.compareAndSet(ScheduleState.NEW, ScheduleState.STARTING)) { + throw new IllegalStateException("Schedule already started or stopped: " + state.get()); + } + + try { + switch (config.getMode()) { + case CRON: + scheduledFuture = taskScheduler.schedule(this::executeGraph, + new CronTrigger(config.getCronExpression())); + break; + case FIXED_DELAY: + scheduledFuture = taskScheduler.scheduleWithFixedDelay(this::executeGraph, + Instant.now().plusMillis(config.getInitialDelay()), Duration.ofMillis(config.getFixedDelay())); + break; + case FIXED_RATE: + scheduledFuture = taskScheduler.scheduleAtFixedRate(this::executeGraph, + Instant.now().plusMillis(config.getInitialDelay()), Duration.ofMillis(config.getFixedRate())); + break; + case ONE_TIME: + scheduledFuture = taskScheduler.schedule(this::executeGraph, + Instant.now().plusMillis(config.getInitialDelay())); + break; + case TRIGGER: + scheduledFuture = taskScheduler.schedule(this::executeGraph, config.getTrigger()); + break; + default: + throw new IllegalStateException("Unexpected value: " + config.getMode()); + } + } + catch (RuntimeException e) { + // A failed start must not leave the task registered or scheduled (#215). + // CAS, not set: if stop() already won the STARTING window it owns the + // terminal state, and FAILED must not clobber its STOPPED. + state.compareAndSet(ScheduleState.STARTING, ScheduleState.FAILED); + unregisterTask(); + throw e; } - switch (config.getMode()) { - case CRON: - scheduledFuture = taskScheduler.schedule(this::executeGraph, - new CronTrigger(config.getCronExpression())); - break; - case FIXED_DELAY: - scheduledFuture = taskScheduler.scheduleWithFixedDelay(this::executeGraph, - Instant.now().plusMillis(config.getInitialDelay()), Duration.ofMillis(config.getFixedDelay())); - break; - case FIXED_RATE: - scheduledFuture = taskScheduler.scheduleAtFixedRate(this::executeGraph, - Instant.now().plusMillis(config.getInitialDelay()), Duration.ofMillis(config.getFixedRate())); - break; - case ONE_TIME: - scheduledFuture = taskScheduler.schedule(this::executeGraph, - Instant.now().plusMillis(config.getInitialDelay())); - break; - case TRIGGER: - scheduledFuture = taskScheduler.schedule(this::executeGraph, config.getTrigger()); - break; - default: - throw new IllegalStateException("Unexpected value: " + config.getMode()); + if (!state.compareAndSet(ScheduleState.STARTING, ScheduleState.RUNNING)) { + // stop() won the race while the future was being published: cancel the late + // publication so it cannot trigger new executions, and finish stopped. + cancelPublishedFuture(); + throw new IllegalStateException("Schedule stopped while starting"); } - started = true; notifyListeners(ScheduleLifecycleListener.ScheduleEvent.STARTED); return this; } @@ -110,16 +153,44 @@ public ScheduledAgentTask start() { * Stop the scheduled execution */ public void stop() { - if (scheduledFuture != null && !scheduledFuture.isCancelled()) { - scheduledFuture.cancel(false); + // One winner per stop: concurrent and repeated stops return idempotently. + if (!transitionToStopping()) { + return; } - stopped = true; + cancelPublishedFuture(); // Unregister from the manager that owns this task. unregisterTask(); log.debug("Stopped and unregistered ScheduledAgentTask with ID: {}", taskId); notifyListeners(ScheduleLifecycleListener.ScheduleEvent.STOPPED); + state.set(ScheduleState.STOPPED); + } + + /** + * Moves a non-terminal state to STOPPING under one CAS winner, so a concurrent or + * repeated stop is a no-op. + * @return true if this caller won the transition and must perform the stop work + */ + private boolean transitionToStopping() { + while (true) { + ScheduleState current = state.get(); + if (current == ScheduleState.STOPPING || current == ScheduleState.STOPPED + || current == ScheduleState.FAILED) { + return false; + } + if (state.compareAndSet(current, ScheduleState.STOPPING)) { + return true; + } + } + } + + /** Cancels the published future if present; safe against a late publication race. */ + private void cancelPublishedFuture() { + ScheduledFuture active = scheduledFuture; + if (active != null && !active.isCancelled()) { + active.cancel(false); + } } /** @@ -222,11 +293,16 @@ private void unregisterTask() { // Getters for monitoring public boolean isStarted() { - return started; + return state.get() != ScheduleState.NEW && state.get() != ScheduleState.FAILED; } public boolean isStopped() { - return stopped; + return state.get() == ScheduleState.STOPPING || state.get() == ScheduleState.STOPPED; + } + + /** Test observation point for the lifecycle state machine. */ + ScheduleState state() { + return state.get(); } /** diff --git a/argi-orchestration/argi-workflow-graph-core/src/test/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTaskAtomicStartStopTest.java b/argi-orchestration/argi-workflow-graph-core/src/test/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTaskAtomicStartStopTest.java new file mode 100644 index 000000000..9b2520df3 --- /dev/null +++ b/argi-orchestration/argi-workflow-graph-core/src/test/java/io/github/agentic/ai/workflow/graph/scheduling/ScheduledAgentTaskAtomicStartStopTest.java @@ -0,0 +1,372 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.workflow.graph.scheduling; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Delayed; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + + +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.Trigger; + +import static io.github.agentic.ai.workflow.graph.scheduling.ScheduledAgentTaskLifecycleTest.graphFailingFirstAttempts; +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** 复现 #215：start/stop 必须是原子状态转换，双 start 只留一份调度，stop 后不得复活。 */ +class ScheduledAgentTaskAtomicStartStopTest { + + private final ScheduledAgentManagerFactory factory = ScheduledAgentManagerFactory.getInstance(); + + private final ScheduledAgentManager originalManager = factory.getManager(); + + private final TestManager manager = new TestManager(); + + /** 恢复测试前的管理器，避免全局 provider 影响其他测试。 */ + @AfterEach + void restoreOriginalManager() { + factory.registerProvider(() -> originalManager); + } + + /** 验证第一个 start 卡在调度发布窗口时，并发的第二个 start 被拒绝且只产生一份调度。 */ + @Test + void aConcurrentSecondStartWhileTheFirstIsPublishingIsRejectedAndSchedulesOnce() throws Exception { + factory.registerProvider(() -> manager); + CountDownLatch scheduleEntered = new CountDownLatch(1); + CountDownLatch releaseSchedule = new CountDownLatch(1); + manager.scheduler.gateNextSchedule(scheduleEntered, releaseSchedule); + final ScheduledAgentTask task = new ScheduledAgentTask(graphFailingFirstAttempts(new AtomicInteger(), 0), + ScheduleConfig.builder().initialDelay(0).build()); + AtomicReference firstFailure = new AtomicReference<>(); + AtomicReference secondFailure = new AtomicReference<>(); + Thread first = startInBackground(task, firstFailure, "first-start"); + Thread second = startInBackground(task, secondFailure, "second-start"); + try { + first.start(); + // 第一个 start 已进入调度发布（旧代码此刻 started 仍为 false）。 + assertTrue(scheduleEntered.await(5, TimeUnit.SECONDS), "first start must be inside schedule"); + second.start(); + second.join(TimeUnit.SECONDS.toMillis(5)); + releaseSchedule.countDown(); + first.join(TimeUnit.SECONDS.toMillis(5)); + + assertAll(() -> assertFalse(first.isAlive(), "first start must finish"), + () -> assertFalse(second.isAlive(), "second start must finish"), + () -> assertNull(firstFailure.get(), "The winning start must succeed"), + () -> assertTrue(secondFailure.get() instanceof IllegalStateException, + "The losing concurrent start must be rejected, got: " + secondFailure.get()), + () -> assertEquals(1, manager.scheduler.scheduleCount(), + "A double start must not publish two schedules")); + } + finally { + releaseSchedule.countDown(); + first.join(TimeUnit.SECONDS.toMillis(5)); + second.join(TimeUnit.SECONDS.toMillis(5)); + } + } + + /** 验证 stop 之后再 start 被拒绝：已停止的任务不得重新进入调度。 */ + @Test + void aStartAfterStopIsRejectedAndPublishesNoSchedule() throws Exception { + factory.registerProvider(() -> manager); + final ScheduledAgentTask task = new ScheduledAgentTask(graphFailingFirstAttempts(new AtomicInteger(), 0), + ScheduleConfig.builder().initialDelay(0).build()); + task.stop(); + assertTrue(task.isStopped()); + + assertThrows(IllegalStateException.class, task::start); + + assertEquals(0, manager.scheduler.scheduleCount(), "A stopped task must not be scheduled again"); + } + + /** 验证 start 调度失败时不遗留注册，异常原样传播。 */ + @Test + void aFailedStartUnregistersTheTaskAndPropagatesTheCause() throws Exception { + factory.registerProvider(() -> manager); + final ScheduledAgentTask task = new ScheduledAgentTask(graphFailingFirstAttempts(new AtomicInteger(), 0), + ScheduleConfig.builder().initialDelay(0).build()); + assertEquals(1, manager.getActiveTaskCount()); + RuntimeException schedulerBroken = new IllegalStateException("scheduler broken"); + manager.scheduler.throwOnSchedule(schedulerBroken); + + RuntimeException thrown = assertThrows(IllegalStateException.class, task::start); + + assertAll(() -> assertEquals(schedulerBroken.getMessage(), thrown.getMessage(), "cause must propagate"), + () -> assertEquals(0, manager.getActiveTaskCount(), + "A failed start must not leave the task registered"), + () -> assertFalse(task.isStarted(), "A failed start is not started"), + () -> assertFalse(task.isStopped(), "A failed start is not stopped either")); + } + + /** 验证 STARTING 窗口内的 stop 取消迟到发布的 future，start 以异常结束且终态停止。 */ + @Test + void aStopDuringStartCancelsTheLateFutureAndFinishesStopped() throws Exception { + factory.registerProvider(() -> manager); + CountDownLatch scheduleEntered = new CountDownLatch(1); + CountDownLatch releaseSchedule = new CountDownLatch(1); + manager.scheduler.gateNextSchedule(scheduleEntered, releaseSchedule); + final ScheduledAgentTask task = new ScheduledAgentTask(graphFailingFirstAttempts(new AtomicInteger(), 0), + ScheduleConfig.builder().initialDelay(0).build()); + AtomicReference startFailure = new AtomicReference<>(); + Thread starter = startInBackground(task, startFailure, "racing-start"); + try { + starter.start(); + assertTrue(scheduleEntered.await(5, TimeUnit.SECONDS), "start must be inside schedule"); + + task.stop(); + releaseSchedule.countDown(); + starter.join(TimeUnit.SECONDS.toMillis(5)); + + assertAll(() -> assertFalse(starter.isAlive(), "start must finish"), + () -> assertTrue(startFailure.get() instanceof IllegalStateException, + "A start losing the race to stop must fail, got: " + startFailure.get()), + () -> assertTrue(task.isStopped(), "The task must end stopped"), + () -> assertEquals(0, manager.getActiveTaskCount(), "The task must be unregistered"), + () -> assertTrue(manager.scheduler.lastFuture().isCancelled(), + "The late-published future must be cancelled, not left live")); + } + finally { + releaseSchedule.countDown(); + starter.join(TimeUnit.SECONDS.toMillis(5)); + } + } + + /** 验证状态机全链转移：NEW→STARTING→RUNNING→STOPPING→STOPPED，started 历史在停止后保留。 */ + @Test + void theLifecycleWalksNewStartingRunningStoppingStopped() throws Exception { + factory.registerProvider(() -> manager); + final ScheduledAgentTask task = new ScheduledAgentTask(graphFailingFirstAttempts(new AtomicInteger(), 0), + ScheduleConfig.builder().initialDelay(0).build()); + assertEquals(ScheduledAgentTask.ScheduleState.NEW, task.state()); + + task.start(); + assertAll(() -> assertEquals(ScheduledAgentTask.ScheduleState.RUNNING, task.state()), + () -> assertTrue(task.isStarted()), () -> assertFalse(task.isStopped())); + + task.stop(); + assertAll(() -> assertEquals(ScheduledAgentTask.ScheduleState.STOPPED, task.state()), + () -> assertTrue(task.isStarted(), "started history is retained after stop"), + () -> assertTrue(task.isStopped())); + } + + private Thread startInBackground(ScheduledAgentTask task, AtomicReference failure, String name) { + Thread worker = new Thread(() -> { + try { + task.start(); + } + catch (Throwable throwable) { + failure.set(throwable); + } + }, name); + return worker; + } + + /** 测试管理器：自带可控 scheduler，任务注册/注销全部记录。 */ + private static final class TestManager implements ScheduledAgentManager { + + final GatedTaskScheduler scheduler = new GatedTaskScheduler(); + + private final Map tasks = new HashMap<>(); + + private final AtomicInteger sequence = new AtomicInteger(); + + @Override + public String registerTask(ScheduledAgentTask task) { + String taskId = "test-task-" + sequence.incrementAndGet(); + tasks.put(taskId, task); + return taskId; + } + + @Override + public boolean unregisterTask(String taskId) { + return tasks.remove(taskId) != null; + } + + @Override + public Optional getTask(String taskId) { + return Optional.ofNullable(tasks.get(taskId)); + } + + @Override + public Set getAllActiveTaskIds() { + return Set.copyOf(tasks.keySet()); + } + + @Override + public int getActiveTaskCount() { + return tasks.size(); + } + + @Override + public TaskScheduler getTaskScheduler() { + return scheduler; + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public void shutdown() { + tasks.clear(); + } + + } + + /** 可控 TaskScheduler：可阻塞 gate、可注入失败，仅统计调度次数。 */ + private static final class GatedTaskScheduler implements TaskScheduler { + + private final AtomicInteger scheduleCount = new AtomicInteger(); + + private final AtomicReference plannedFailure = new AtomicReference<>(); + + private volatile CountDownLatch scheduleEntered; + + private volatile CountDownLatch releaseSchedule; + + private volatile ScheduledFuture lastFuture; + + private void gateNextSchedule(CountDownLatch entered, CountDownLatch release) { + this.scheduleEntered = entered; + this.releaseSchedule = release; + } + + private void throwOnSchedule(RuntimeException cause) { + plannedFailure.set(cause); + } + + private int scheduleCount() { + return scheduleCount.get(); + } + + private ScheduledFuture lastFuture() { + return lastFuture; + } + + private ScheduledFuture submit() { + scheduleCount.incrementAndGet(); + RuntimeException planned = plannedFailure.get(); + if (planned != null) { + throw planned; + } + CountDownLatch entered = this.scheduleEntered; + if (entered != null) { + entered.countDown(); + try { + releaseSchedule.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + ScheduledFuture future = new StubScheduledFuture(); + lastFuture = future; + return future; + } + + @Override + public ScheduledFuture schedule(Runnable task, Trigger trigger) { + return submit(); + } + + @Override + public ScheduledFuture schedule(Runnable task, Instant startTime) { + return submit(); + } + + @Override + public ScheduledFuture scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) { + return submit(); + } + + @Override + public ScheduledFuture scheduleAtFixedRate(Runnable task, Duration period) { + return submit(); + } + + @Override + public ScheduledFuture scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) { + return submit(); + } + + @Override + public ScheduledFuture scheduleWithFixedDelay(Runnable task, Duration delay) { + return submit(); + } + + } + + private static final class StubScheduledFuture implements ScheduledFuture { + + private volatile boolean cancelled; + + @Override + public long getDelay(TimeUnit unit) { + return 0; + } + + @Override + public int compareTo(Delayed other) { + return 0; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + cancelled = true; + return true; + } + + @Override + public boolean isCancelled() { + return cancelled; + } + + @Override + public boolean isDone() { + return cancelled; + } + + @Override + public Object get() { + return null; + } + + @Override + public Object get(long timeout, TimeUnit unit) { + return null; + } + + } + +}