feat: M28 支持工作流多入边汇聚模式

- 持久化 Quartz refire 状态并收口运行时启动关闭顺序

- 增加批量 Trigger 获取配置、校验与回归测试

easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java

@@ -634,7 +634,8 @@ public class Chain {
                                         NodeStateField
                                                 .EXECUTION_ATTEMPT_KEY);
                             }
-                            if (node.getCondition() == null) {
+                            if (node.getJoinMode() == NodeJoinMode.ANY
+                                    && node.getCondition() == null) {
                                 s.recordTrigger(triggerEdgeId);
                                 fields.add(NodeStateField.TRIGGER_COUNT);
                                 fields.add(NodeStateField.TRIGGER_EDGE_IDS);
@@ -795,7 +796,7 @@ public class Chain {
 
     private boolean shouldSkipNode(Node node, String edgeId) {
         NodeCondition condition = node.getCondition();
-        if (condition == null) {
+        if (node.getJoinMode() == NodeJoinMode.ANY && condition == null) {
             return false;
         }
         return executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS, () -> {
@@ -805,7 +806,11 @@ public class Chain {
             });
 
             Map<String, Object> prevResult = Collections.emptyMap();
-            boolean shouldSkipNode = !condition.check(this, newState, prevResult);
+            boolean joinPending = node.getJoinMode() == NodeJoinMode.ALL
+                    && !newState.isUpstreamFullyExecuted();
+            boolean shouldSkipNode = joinPending
+                    || (condition != null
+                    && !condition.check(this, newState, prevResult));
             if (shouldSkipNode) {
                 updateStateSafely(state -> {
                     return state.addUncheckedNodeId(node.id)

easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Node.java

@@ -42,6 +42,7 @@ public abstract class Node implements Serializable {
 
     protected NodeCondition condition;
     protected NodeValidator validator;
+    protected NodeJoinMode joinMode = NodeJoinMode.ANY;
 
     // 循环执行相关属性
     protected boolean loopEnable = false;           // 是否启用循环执行
@@ -70,6 +71,10 @@ public abstract class Node implements Serializable {
     }
 
     public void setParentId(String parentId) {
+        if (StringUtil.hasText(parentId) && getJoinMode() == NodeJoinMode.ALL) {
+            throw new IllegalArgumentException(
+                    "joinMode 'all' is not supported for loop child nodes");
+        }
         this.parentId = parentId;
     }
 
@@ -121,6 +126,31 @@ public abstract class Node implements Serializable {
         this.validator = validator;
     }
 
+    /**
+     * 获取节点的直接入边汇聚模式。
+     *
+     * <p>旧序列化对象缺少该字段时返回 {@link NodeJoinMode#ANY}。</p>
+     *
+     * @return 汇聚模式
+     */
+    public NodeJoinMode getJoinMode() {
+        return joinMode == null ? NodeJoinMode.ANY : joinMode;
+    }
+
+    /**
+     * 设置节点的直接入边汇聚模式。
+     *
+     * @param joinMode 汇聚模式
+     */
+    public void setJoinMode(NodeJoinMode joinMode) {
+        NodeJoinMode resolved = joinMode == null ? NodeJoinMode.ANY : joinMode;
+        if (resolved == NodeJoinMode.ALL && StringUtil.hasText(parentId)) {
+            throw new IllegalArgumentException(
+                    "joinMode 'all' is not supported for loop child nodes");
+        }
+        this.joinMode = resolved;
+    }
+
 //    protected void addOutwardEdge(Edge edge) {
 //        if (this.outwardEdges == null) {
 //            this.outwardEdges = new ArrayList<>();

easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/NodeJoinMode.java

@@ -0,0 +1,60 @@
+/**
+ * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
+ * <p>
+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.gnu.org/licenses/lgpl-3.0.txt
+ * <p>
+ * 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 com.easyagents.flow.core.chain;
+
+import java.util.Locale;
+
+/**
+ * 多入边节点的触发汇聚模式。
+ */
+public enum NodeJoinMode {
+
+    /** 任意一条直接入边到达即可执行。 */
+    ANY("any"),
+    /** 全部直接入边到达后才执行。 */
+    ALL("all");
+
+    private final String value;
+
+    NodeJoinMode(String value) {
+        this.value = value;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    /**
+     * 按工作流 JSON 值解析汇聚模式。
+     *
+     * @param value 配置值
+     * @return 汇聚模式
+     * @throws IllegalArgumentException 配置为空或不受支持
+     */
+    public static NodeJoinMode ofValue(String value) {
+        if (value == null || value.trim().isEmpty()) {
+            throw new IllegalArgumentException("joinMode must be 'any' or 'all'");
+        }
+        String normalized = value.trim().toLowerCase(Locale.ROOT);
+        for (NodeJoinMode mode : values()) {
+            if (mode.value.equals(normalized)) {
+                return mode;
+            }
+        }
+        throw new IllegalArgumentException(
+                "Unsupported joinMode: " + value + "; expected 'any' or 'all'");
+    }
+}

easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java

@@ -21,6 +21,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.easyagents.flow.core.chain.DataType;
 import com.easyagents.flow.core.chain.JsCodeCondition;
 import com.easyagents.flow.core.chain.Node;
+import com.easyagents.flow.core.chain.NodeJoinMode;
 import com.easyagents.flow.core.chain.Parameter;
 import com.easyagents.flow.core.chain.RefType;
 import com.easyagents.flow.core.node.BaseNode;
@@ -123,6 +124,10 @@ public abstract class BaseNodeParser<T extends BaseNode> implements NodeParser<T
 
             if (!data.isEmpty()) {
 
+                if (data.containsKey("joinMode")) {
+                    node.setJoinMode(NodeJoinMode.ofValue(data.getString("joinMode")));
+                }
+
                 addParameters(node, data);
                 addOutputDefs(node, data);
 

easy-agents-flow/src/test/java/com/easyagents/flow/core/test/NodeJoinModeTest.java

@@ -0,0 +1,407 @@
+package com.easyagents.flow.core.test;
+
+import com.alibaba.fastjson.JSONObject;
+import com.easyagents.flow.core.chain.Chain;
+import com.easyagents.flow.core.chain.ChainDefinition;
+import com.easyagents.flow.core.chain.ChainState;
+import com.easyagents.flow.core.chain.ChainStatus;
+import com.easyagents.flow.core.chain.Edge;
+import com.easyagents.flow.core.chain.Node;
+import com.easyagents.flow.core.chain.NodeCondition;
+import com.easyagents.flow.core.chain.NodeJoinMode;
+import com.easyagents.flow.core.chain.NodeState;
+import com.easyagents.flow.core.chain.Parameter;
+import com.easyagents.flow.core.chain.RefType;
+import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository;
+import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
+import com.easyagents.flow.core.chain.runtime.ChainExecutor;
+import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore;
+import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
+import com.easyagents.flow.core.node.BaseNode;
+import com.easyagents.flow.core.node.EndNode;
+import com.easyagents.flow.core.node.StartNode;
+import com.easyagents.flow.core.parser.ChainParser;
+import com.easyagents.flow.core.parser.impl.EndNodeParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+
+/**
+ * 验证普通节点的直接入边汇聚模式。
+ */
+public class NodeJoinModeTest {
+
+    @Test
+    public void shouldParseJoinModeAndRejectInvalidValues() {
+        ChainParser parser = ChainParser.builder()
+                .withDefaultParsers(true)
+                .build();
+
+        Assert.assertEquals(
+                NodeJoinMode.ANY,
+                parseEndNode(parser, null, null).getJoinMode());
+        Assert.assertEquals(
+                NodeJoinMode.ALL,
+                parseEndNode(parser, "all", null).getJoinMode());
+        Assert.assertEquals(
+                NodeJoinMode.ANY,
+                parseEndNode(parser, "ANY", null).getJoinMode());
+
+        assertInvalidJoinMode(() -> parseEndNode(parser, "first", null));
+        assertInvalidJoinMode(() -> parseEndNode(parser, "", null));
+        assertInvalidJoinMode(() -> parseEndNode(parser, "all", "loop"));
+
+        ProbeJoinNode node = new ProbeJoinNode(new AtomicInteger());
+        node.setJoinMode(NodeJoinMode.ALL);
+        assertInvalidJoinMode(() -> node.setParentId("loop"));
+    }
+
+    @Test
+    public void shouldWaitForEveryInboundEdgeBeforeExecutingAndCheckingCondition()
+            throws Exception {
+        JoinFixture fixture = createJoinFixture(NodeJoinMode.ALL, true);
+        String instanceId = null;
+        try {
+            instanceId = fixture.executor.executeAsync(
+                    fixture.definition.getId(), Collections.emptyMap());
+            Assert.assertTrue(fixture.branchACompleted.await(2, TimeUnit.SECONDS));
+            Assert.assertTrue(fixture.branchBStarted.await(2, TimeUnit.SECONDS));
+
+            String currentInstanceId = instanceId;
+            await(() -> hasTriggerEdge(
+                    fixture.nodeStateRepository,
+                    currentInstanceId,
+                    "join",
+                    "a-join"));
+            Assert.assertEquals(0, fixture.joinExecutions.get());
+            Assert.assertEquals(0, fixture.conditionChecks.get());
+
+            fixture.releaseBranchB.countDown();
+            ChainState finalState = awaitTerminal(
+                    fixture.chainStateRepository, instanceId);
+
+            Assert.assertEquals(1, fixture.joinExecutions.get());
+            Assert.assertEquals(1, fixture.conditionChecks.get());
+            Assert.assertEquals("A+B", finalState.getExecuteResult().get("combined"));
+            Assert.assertEquals(Boolean.TRUE, finalState.getExecuteResult().get("sawBoth"));
+        } finally {
+            fixture.releaseBranchB.countDown();
+            fixture.scheduler.shutdown();
+        }
+    }
+
+    @Test
+    public void shouldKeepAnyModeFirstArrivalBehavior() throws Exception {
+        JoinFixture fixture = createJoinFixture(NodeJoinMode.ANY, false);
+        try {
+            String instanceId = fixture.executor.executeAsync(
+                    fixture.definition.getId(), Collections.emptyMap());
+            Assert.assertTrue(fixture.branchACompleted.await(2, TimeUnit.SECONDS));
+            Assert.assertTrue(fixture.branchBStarted.await(2, TimeUnit.SECONDS));
+
+            await(() -> fixture.joinExecutions.get() > 0);
+            ChainState finalState = awaitTerminal(
+                    fixture.chainStateRepository, instanceId);
+
+            Assert.assertEquals(1, fixture.joinExecutions.get());
+            Assert.assertEquals(Boolean.FALSE, finalState.getExecuteResult().get("sawBoth"));
+        } finally {
+            fixture.releaseBranchB.countDown();
+            fixture.scheduler.shutdown();
+        }
+    }
+
+    @Test
+    public void shouldKeepDefaultAnyRetryAndLoopBehavior() throws Exception {
+        ScheduledExecutorService schedulerPool =
+                Executors.newSingleThreadScheduledExecutor();
+        ExecutorService workerPool = Executors.newFixedThreadPool(3);
+        TriggerScheduler scheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(), schedulerPool, workerPool, 1000L);
+        AtomicInteger executions = new AtomicInteger();
+        ChainDefinition definition = new ChainDefinition();
+        definition.setId("join-mode-retry-loop");
+        StartNode start = new StartNode();
+        start.setId("start");
+        RetryLoopNode worker = new RetryLoopNode(executions);
+        worker.setId("worker");
+        worker.setRetryEnable(true);
+        worker.setMaxRetryCount(1);
+        worker.setRetryIntervalMs(0L);
+        worker.setLoopEnable(true);
+        worker.setMaxLoopCount(2);
+        worker.setLoopIntervalMs(0L);
+        EndNode end = endNode("worker", "count", "count");
+        definition.addNode(start);
+        definition.addNode(worker);
+        definition.addNode(end);
+        definition.addEdge(edge("start-worker", "start", "worker"));
+        definition.addEdge(edge("worker-end", "worker", "end"));
+
+        ChainExecutor executor = new ChainExecutor(
+                ignored -> definition,
+                new InMemoryChainStateRepository(),
+                new InMemoryNodeStateRepository(),
+                scheduler);
+        try {
+            Map<String, Object> result = executor.execute(
+                    definition.getId(), Collections.emptyMap(), 5L, TimeUnit.SECONDS);
+
+            Assert.assertEquals(NodeJoinMode.ANY, worker.getJoinMode());
+            Assert.assertEquals(3, executions.get());
+            Assert.assertEquals(3, result.get("count"));
+        } finally {
+            scheduler.shutdown();
+        }
+    }
+
+    private JoinFixture createJoinFixture(
+            NodeJoinMode joinMode, boolean withCondition) {
+        JoinFixture fixture = new JoinFixture();
+        fixture.schedulerPool = Executors.newScheduledThreadPool(2);
+        fixture.workerPool = Executors.newFixedThreadPool(4);
+        fixture.scheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(),
+                fixture.schedulerPool,
+                fixture.workerPool,
+                1000L);
+        fixture.chainStateRepository = new InMemoryChainStateRepository();
+        fixture.nodeStateRepository = new InMemoryNodeStateRepository();
+        fixture.definition = new ChainDefinition();
+        fixture.definition.setId("join-mode-" + joinMode.getValue());
+
+        StartNode start = new StartNode();
+        start.setId("start");
+        BranchNode branchA = new BranchNode(
+                "A", fixture.branchACompleted, null, null);
+        branchA.setId("a");
+        BranchNode branchB = new BranchNode(
+                "B", null, fixture.branchBStarted, fixture.releaseBranchB);
+        branchB.setId("b");
+        ProbeJoinNode join = new ProbeJoinNode(fixture.joinExecutions);
+        join.setId("join");
+        join.setJoinMode(joinMode);
+        if (withCondition) {
+            join.setCondition(new BothOutputsCondition(fixture.conditionChecks));
+        }
+        EndNode end = endNode("join", "combined", "combined");
+        end.addOutputDef(outputRef("sawBoth", "join.sawBoth"));
+
+        fixture.definition.addNode(start);
+        fixture.definition.addNode(branchA);
+        fixture.definition.addNode(branchB);
+        fixture.definition.addNode(join);
+        fixture.definition.addNode(end);
+        fixture.definition.addEdge(edge("start-a", "start", "a"));
+        fixture.definition.addEdge(edge("start-b", "start", "b"));
+        fixture.definition.addEdge(edge("a-join", "a", "join"));
+        fixture.definition.addEdge(edge("b-join", "b", "join"));
+        fixture.definition.addEdge(edge("join-end", "join", "end"));
+
+        fixture.executor = new ChainExecutor(
+                ignored -> fixture.definition,
+                fixture.chainStateRepository,
+                fixture.nodeStateRepository,
+                fixture.scheduler);
+        return fixture;
+    }
+
+    private Node parseEndNode(
+            ChainParser parser, String joinMode, String parentId) {
+        JSONObject data = new JSONObject();
+        if (joinMode != null) {
+            data.put("joinMode", joinMode);
+        }
+        JSONObject nodeJson = new JSONObject();
+        nodeJson.put("id", "end");
+        nodeJson.put("type", "endNode");
+        nodeJson.put("parentId", parentId);
+        nodeJson.put("data", data);
+        return new EndNodeParser().parse(
+                nodeJson, new JSONObject(), parser);
+    }
+
+    private void assertInvalidJoinMode(Runnable action) {
+        try {
+            action.run();
+            Assert.fail("Expected invalid join mode");
+        } catch (IllegalArgumentException expected) {
+            Assert.assertTrue(expected.getMessage().contains("joinMode"));
+        }
+    }
+
+    private static boolean hasTriggerEdge(
+            InMemoryNodeStateRepository repository,
+            String instanceId,
+            String nodeId,
+            String edgeId) {
+        NodeState state = repository.load(instanceId, nodeId);
+        return state != null && state.getTriggerEdgeIds().contains(edgeId);
+    }
+
+    private static ChainState awaitTerminal(
+            InMemoryChainStateRepository repository,
+            String instanceId) throws Exception {
+        await(() -> {
+            ChainState state = repository.load(instanceId);
+            return state != null
+                    && state.getStatus() != null
+                    && state.getStatus().isTerminal();
+        });
+        ChainState state = repository.load(instanceId);
+        Assert.assertEquals(ChainStatus.SUCCEEDED, state.getStatus());
+        return state;
+    }
+
+    private static void await(BooleanSupplier condition) throws Exception {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3L);
+        while (!condition.getAsBoolean() && System.nanoTime() < deadline) {
+            Thread.sleep(10L);
+        }
+        Assert.assertTrue("condition was not met before timeout", condition.getAsBoolean());
+    }
+
+    private static EndNode endNode(
+            String sourceNodeId, String sourceName, String outputName) {
+        EndNode end = new EndNode();
+        end.setId("end");
+        end.addOutputDef(outputRef(
+                outputName, sourceNodeId + "." + sourceName));
+        return end;
+    }
+
+    private static Parameter outputRef(String name, String ref) {
+        Parameter parameter = new Parameter();
+        parameter.setName(name);
+        parameter.setRef(ref);
+        parameter.setRefType(RefType.REF);
+        return parameter;
+    }
+
+    private static Edge edge(String id, String source, String target) {
+        Edge edge = new Edge();
+        edge.setId(id);
+        edge.setSource(source);
+        edge.setTarget(target);
+        return edge;
+    }
+
+    private static final class BranchNode extends BaseNode {
+        private final String value;
+        private final CountDownLatch completed;
+        private final CountDownLatch started;
+        private final CountDownLatch release;
+
+        private BranchNode(
+                String value,
+                CountDownLatch completed,
+                CountDownLatch started,
+                CountDownLatch release) {
+            this.value = value;
+            this.completed = completed;
+            this.started = started;
+            this.release = release;
+        }
+
+        @Override
+        public Map<String, Object> execute(Chain chain) {
+            if (started != null) {
+                started.countDown();
+            }
+            if (release != null) {
+                try {
+                    if (!release.await(3L, TimeUnit.SECONDS)) {
+                        throw new IllegalStateException("branch release timed out");
+                    }
+                } catch (InterruptedException error) {
+                    Thread.currentThread().interrupt();
+                    throw new IllegalStateException("branch interrupted", error);
+                }
+            }
+            if (completed != null) {
+                completed.countDown();
+            }
+            return Collections.singletonMap("value", value);
+        }
+    }
+
+    private static final class ProbeJoinNode extends BaseNode {
+        private final AtomicInteger executions;
+
+        private ProbeJoinNode(AtomicInteger executions) {
+            this.executions = executions;
+        }
+
+        @Override
+        public Map<String, Object> execute(Chain chain) {
+            executions.incrementAndGet();
+            Object a = chain.getExecutionState().getMemory().get("a.value");
+            Object b = chain.getExecutionState().getMemory().get("b.value");
+            Map<String, Object> result = new HashMap<>();
+            result.put("combined", String.valueOf(a) + "+" + String.valueOf(b));
+            result.put("sawBoth", a != null && b != null);
+            return result;
+        }
+    }
+
+    private static final class BothOutputsCondition implements NodeCondition {
+        private static final long serialVersionUID = 1L;
+        private final AtomicInteger checks;
+
+        private BothOutputsCondition(AtomicInteger checks) {
+            this.checks = checks;
+        }
+
+        @Override
+        public boolean check(
+                Chain chain,
+                NodeState context,
+                Map<String, Object> executeResult) {
+            checks.incrementAndGet();
+            Map<String, Object> memory = chain.getExecutionState().getMemory();
+            return memory.containsKey("a.value") && memory.containsKey("b.value");
+        }
+    }
+
+    private static final class RetryLoopNode extends BaseNode {
+        private final AtomicInteger executions;
+
+        private RetryLoopNode(AtomicInteger executions) {
+            this.executions = executions;
+        }
+
+        @Override
+        public Map<String, Object> execute(Chain chain) {
+            int count = executions.incrementAndGet();
+            if (count == 1) {
+                throw new IllegalStateException("retry once");
+            }
+            return Collections.singletonMap("count", count);
+        }
+    }
+
+    private static final class JoinFixture {
+        private final CountDownLatch branchACompleted = new CountDownLatch(1);
+        private final CountDownLatch branchBStarted = new CountDownLatch(1);
+        private final CountDownLatch releaseBranchB = new CountDownLatch(1);
+        private final AtomicInteger joinExecutions = new AtomicInteger();
+        private final AtomicInteger conditionChecks = new AtomicInteger();
+        private ScheduledExecutorService schedulerPool;
+        private ExecutorService workerPool;
+        private TriggerScheduler scheduler;
+        private InMemoryChainStateRepository chainStateRepository;
+        private InMemoryNodeStateRepository nodeStateRepository;
+        private ChainDefinition definition;
+        private ChainExecutor executor;
+    }
+}

easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java

@@ -0,0 +1,53 @@
+package com.easyagents.scheduler;
+
+import java.time.Duration;
+
+/**
+ * 请求调度提供方持久化重试同一次逻辑触发的异常。
+ *
+ * <p>仅用于尚未产生业务副作用、且调用方必须保证至少一次登记的短 Handler。
+ * {@code maxRefires=0} 表示不限制持久化重试次数；Provider 正常情况下应释放当前执行线程，
+ * 并通过持久化的延迟触发保留原 fire 上下文。仅当 JobStore 无法写入持久重试时，
+ * Provider 可按相同退避暂时占用当前线程并原地重试，以避免正常完成造成 fire 丢失。</p>
+ */
+public class ScheduleRefireException extends RuntimeException {
+
+    private static final int DEFAULT_MAX_REFIRES = 0;
+    private static final Duration DEFAULT_BASE_DELAY = Duration.ofMillis(250);
+    private final int maxRefires;
+    private final Duration baseDelay;
+
+    public ScheduleRefireException(String message, Throwable cause) {
+        this(message, cause, DEFAULT_MAX_REFIRES, DEFAULT_BASE_DELAY);
+    }
+
+    public ScheduleRefireException(String message, Throwable cause,
+                                   int maxRefires, Duration baseDelay) {
+        super(requireMessage(message), cause);
+        if (maxRefires < 0) throw new IllegalArgumentException("maxRefires must not be negative");
+        if (baseDelay == null || baseDelay.isZero() || baseDelay.isNegative()) {
+            throw new IllegalArgumentException("baseDelay must be positive");
+        }
+        this.maxRefires = maxRefires;
+        this.baseDelay = baseDelay;
+    }
+
+    /**
+     * @return 最大重新执行次数，0 表示不限制
+     */
+    public int maxRefires() {
+        return maxRefires;
+    }
+
+    public Duration delayFor(int refireNumber) {
+        int shift = Math.min(4, Math.max(0, refireNumber - 1));
+        return baseDelay.multipliedBy(1L << shift);
+    }
+
+    private static String requireMessage(String message) {
+        if (message == null || message.isBlank()) {
+            throw new IllegalArgumentException("message must not be blank");
+        }
+        return message;
+    }
+}

easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java

@@ -29,6 +29,9 @@ abstract class AbstractDispatchJob implements InterruptableJob {
                 throw new JobExecutionException("easy-agents scheduler runtime is not available");
             }
             quartzRuntime.execute(context);
+        } catch (JobExecutionException exception) {
+            // JobExecutionException 继承 SchedulerException，必须保留 refire 等控制语义。
+            throw exception;
         } catch (SchedulerException exception) {
             throw new JobExecutionException("failed to access scheduler runtime", exception, false);
         } finally {

easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java

@@ -7,9 +7,16 @@ import com.easyagents.scheduler.ScheduleExecutionListener;
 import com.easyagents.scheduler.ScheduleFireContext;
 import com.easyagents.scheduler.ScheduleHandler;
 import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
 import org.quartz.JobDataMap;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.quartz.ObjectAlreadyExistsException;
+import org.quartz.SchedulerException;
+import org.quartz.SimpleScheduleBuilder;
+import org.quartz.Trigger;
+import org.quartz.TriggerBuilder;
+import org.quartz.TriggerKey;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -19,6 +26,10 @@ import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.UUID;
+import java.util.concurrent.locks.LockSupport;
 
 /**
  * 当前 Scheduler 节点的 Handler 和监听器运行时。
@@ -55,15 +66,27 @@ final class QuartzRuntime {
     void execute(JobExecutionContext quartzContext) throws JobExecutionException {
         JobDataMap data = quartzContext.getMergedJobDataMap();
         ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(data);
-        Instant actualFireTime = toInstant(quartzContext.getFireTime(), Instant.now());
+        Instant observedActualFireTime = toInstant(quartzContext.getFireTime(), Instant.now());
+        Instant scheduledFireTime = instantValue(
+            data,
+            QuartzScheduleMapper.KEY_RETRY_SCHEDULED_FIRE_TIME,
+            toInstant(quartzContext.getScheduledFireTime(), observedActualFireTime)
+        );
+        Instant actualFireTime = instantValue(
+            data,
+            QuartzScheduleMapper.KEY_RETRY_ACTUAL_FIRE_TIME,
+            observedActualFireTime
+        );
         ScheduleFireContext context = new ScheduleFireContext(
             definition.id(),
             definition.handlerCode(),
-            toInstant(quartzContext.getScheduledFireTime(), actualFireTime),
+            scheduledFireTime,
             actualFireTime,
-            quartzContext.getFireInstanceId(),
+            stringValue(data, QuartzScheduleMapper.KEY_RETRY_FIRE_INSTANCE_ID,
+                quartzContext.getFireInstanceId()),
             stringValue(data, QuartzScheduleMapper.KEY_INVOCATION),
-            quartzContext.isRecovering(),
+            booleanValue(data, QuartzScheduleMapper.KEY_RETRY_RECOVERING,
+                quartzContext.isRecovering()),
             definition.parameters()
         );
         ScheduleHandler handler = handlers.get(definition.handlerCode());
@@ -82,10 +105,80 @@ final class QuartzRuntime {
             notifySucceeded(context, elapsed(startedAt));
         } catch (Exception exception) {
             notifyFailed(context, elapsed(startedAt), exception);
+            if (exception instanceof ScheduleRefireException retry) {
+                int attempt = Math.max(
+                    intValue(data, QuartzScheduleMapper.KEY_RETRY_ATTEMPT, 0),
+                    quartzContext.getRefireCount()
+                );
+                if (retry.maxRefires() == 0 || attempt < retry.maxRefires()) {
+                    try {
+                        persistRetry(quartzContext, context, retry, attempt + 1);
+                        return;
+                    } catch (SchedulerException retryFailure) {
+                        exception.addSuppressed(retryFailure);
+                        log.error("Failed to persist schedule handler retry: scheduleId={}, attempt={}",
+                            context.scheduleId(), attempt + 1, retryFailure);
+                        if (canRefireInPlace(quartzContext, retry, attempt + 1)) {
+                            // JobStore 暂时不可写时，正常完成原 fire 会造成登记丢失。
+                            // 仅在这条降级路径短时占用当前 worker；一旦持久重试落库即释放。
+                            throw new JobExecutionException(exception, true);
+                        }
+                    }
+                } else {
+                    log.error("Schedule handler retry limit exhausted: scheduleId={}, retries={}",
+                        context.scheduleId(), retry.maxRefires(), exception);
+                }
+            }
+            // 不在 Quartz worker 内无限 refire。持久化重试失败时保留原异常，
+            // 由 requestRecovery 和集群故障恢复处理未完成的 fired trigger。
             throw new JobExecutionException(exception, false);
         }
     }
 
+    private static boolean canRefireInPlace(
+        JobExecutionContext context,
+        ScheduleRefireException retry,
+        int attempt
+    ) {
+        LockSupport.parkNanos(retry.delayFor(attempt).toNanos());
+        if (Thread.currentThread().isInterrupted()) return false;
+        try {
+            return !context.getScheduler().isShutdown();
+        } catch (SchedulerException exception) {
+            return false;
+        }
+    }
+
+    private static void persistRetry(
+        JobExecutionContext quartzContext,
+        ScheduleFireContext context,
+        ScheduleRefireException retry,
+        int attempt
+    ) throws SchedulerException {
+        String source = context.scheduleId() + "|" + context.fireInstanceId() + "|"
+            + String.valueOf(context.invocationId()) + "|" + attempt;
+        String retryName = "retry-" + UUID.nameUUIDFromBytes(
+            source.getBytes(StandardCharsets.UTF_8));
+        TriggerKey retryKey = new TriggerKey(
+            retryName,
+            QuartzScheduleMapper.GROUP_PREFIX + "retry." + context.scheduleId().namespace()
+        );
+        Trigger trigger = TriggerBuilder.newTrigger()
+            .withIdentity(retryKey)
+            .forJob(quartzContext.getJobDetail().getKey())
+            .usingJobData(QuartzScheduleMapper.retryData(context, attempt))
+            .startAt(Date.from(Instant.now().plus(retry.delayFor(attempt))))
+            .withSchedule(SimpleScheduleBuilder.simpleSchedule()
+                .withRepeatCount(0)
+                .withMisfireHandlingInstructionFireNow())
+            .build();
+        try {
+            quartzContext.getScheduler().scheduleJob(trigger);
+        } catch (ObjectAlreadyExistsException ignored) {
+            // 同一原始 fire/attempt 的确定性 key 已落库，即视为持久化成功。
+        }
+    }
+
     /**
      * 向监听器发布 Misfire 事件。
      *
@@ -175,4 +268,24 @@ final class QuartzRuntime {
         Object value = data.get(key);
         return value == null ? null : value.toString();
     }
+
+    private static String stringValue(JobDataMap data, String key, String fallback) {
+        String value = stringValue(data, key);
+        return value == null ? fallback : value;
+    }
+
+    private static Instant instantValue(JobDataMap data, String key, Instant fallback) {
+        String value = stringValue(data, key);
+        return value == null ? fallback : Instant.ofEpochMilli(Long.parseLong(value));
+    }
+
+    private static boolean booleanValue(JobDataMap data, String key, boolean fallback) {
+        String value = stringValue(data, key);
+        return value == null ? fallback : Boolean.parseBoolean(value);
+    }
+
+    private static int intValue(JobDataMap data, String key, int fallback) {
+        String value = stringValue(data, key);
+        return value == null ? fallback : Integer.parseInt(value);
+    }
 }

easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java

@@ -7,6 +7,7 @@ import com.easyagents.scheduler.OnceSchedulePlan;
 import com.easyagents.scheduler.ScheduleDefinition;
 import com.easyagents.scheduler.ScheduleErrorCode;
 import com.easyagents.scheduler.ScheduleException;
+import com.easyagents.scheduler.ScheduleFireContext;
 import com.easyagents.scheduler.ScheduleId;
 import com.easyagents.scheduler.SchedulePlan;
 import org.quartz.CronScheduleBuilder;
@@ -47,6 +48,11 @@ final class QuartzScheduleMapper {
     static final String KEY_DESCRIPTION = "ea.description";
     static final String KEY_INVOCATION = "ea.invocationId";
     static final String KEY_IMMEDIATE_PARAMETER_SNAPSHOT = "ea.immediateParameterSnapshot";
+    static final String KEY_RETRY_ATTEMPT = "ea.retry.attempt";
+    static final String KEY_RETRY_SCHEDULED_FIRE_TIME = "ea.retry.scheduledFireTime";
+    static final String KEY_RETRY_ACTUAL_FIRE_TIME = "ea.retry.actualFireTime";
+    static final String KEY_RETRY_FIRE_INSTANCE_ID = "ea.retry.fireInstanceId";
+    static final String KEY_RETRY_RECOVERING = "ea.retry.recovering";
     static final String PARAMETER_PREFIX = "ea.parameter.";
     static final String IMMEDIATE_PARAMETER_PREFIX = "ea.immediateParameter.";
     static final String GROUP_PREFIX = "ea.scheduler.";
@@ -153,6 +159,30 @@ final class QuartzScheduleMapper {
         return data;
     }
 
+    /**
+     * 固化持久化重试所需的原始 fire 上下文。
+     */
+    static JobDataMap retryData(ScheduleFireContext context, int attempt) {
+        JobDataMap data = identityData(context.scheduleId());
+        // Trigger 数据覆盖 JobDetail；任务定义被 replace 后，既有 fire 的重试仍须
+        // 派发给首次触发时的 Handler，而不是意外切换到新 Handler。
+        data.put(KEY_HANDLER, context.handlerCode());
+        data.put(KEY_RETRY_ATTEMPT, Integer.toString(attempt));
+        data.put(KEY_RETRY_SCHEDULED_FIRE_TIME,
+            Long.toString(context.scheduledFireTime().toEpochMilli()));
+        data.put(KEY_RETRY_ACTUAL_FIRE_TIME,
+            Long.toString(context.actualFireTime().toEpochMilli()));
+        data.put(KEY_RETRY_FIRE_INSTANCE_ID, context.fireInstanceId());
+        data.put(KEY_RETRY_RECOVERING, Boolean.toString(context.recovering()));
+        if (context.invocationId() != null) {
+            data.put(KEY_INVOCATION, context.invocationId());
+        }
+        // 重试必须沿用首次 fire 的参数快照，不能读取期间被替换的新定义参数。
+        data.put(KEY_IMMEDIATE_PARAMETER_SNAPSHOT, Boolean.TRUE.toString());
+        putParameters(data, context.parameters(), IMMEDIATE_PARAMETER_PREFIX);
+        return data;
+    }
+
     /**
      * 从持久 JobData 恢复公共定义。
      *

easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java

@@ -335,7 +335,8 @@ public final class QuartzScheduleService implements ScheduleService, AutoCloseab
         }
         try {
             if (!scheduler.isShutdown()) {
-                // false 保证 close 本身有界；Quartz 会向内部 InterruptableJob 发送中断。
+                // Factory 启用 interruptJobsOnShutdown，Quartz 会先中断内部
+                // InterruptableJob；false 只表示不再无界等待忽略中断的 Handler。
                 scheduler.shutdown(false);
             }
         } catch (SchedulerException exception) {

easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java

@@ -10,6 +10,8 @@ package com.easyagents.scheduler.quartz;
  * @param clustered 是否启用 JDBC 集群
  * @param threadCount Quartz Worker 线程数
  * @param threadPriority Quartz Worker 线程优先级
+ * @param batchTriggerAcquisitionMaxCount 单次批量获取 Trigger 的最大数量
+ * @param batchTriggerAcquisitionFireAheadTimeWindowMillis 可提前纳入批量的时间窗口，单位毫秒
  * @param clusterCheckinIntervalMillis 集群心跳间隔，单位毫秒
  * @param misfireThresholdMillis Misfire 判定阈值，单位毫秒
  * @param waitForJobsToCompleteOnShutdown 关闭时是否等待运行中任务完成
@@ -24,6 +26,8 @@ public record QuartzSchedulerConfig(
     boolean clustered,
     int threadCount,
     int threadPriority,
+    int batchTriggerAcquisitionMaxCount,
+    long batchTriggerAcquisitionFireAheadTimeWindowMillis,
     long clusterCheckinIntervalMillis,
     long misfireThresholdMillis,
     boolean waitForJobsToCompleteOnShutdown,
@@ -55,6 +59,18 @@ public record QuartzSchedulerConfig(
         if (threadPriority < Thread.MIN_PRIORITY || threadPriority > Thread.MAX_PRIORITY) {
             throw new IllegalArgumentException("threadPriority must be between 1 and 10");
         }
+        if (batchTriggerAcquisitionMaxCount < 1
+                || batchTriggerAcquisitionMaxCount > threadCount) {
+            throw new IllegalArgumentException(
+                "batchTriggerAcquisitionMaxCount must be between 1 and threadCount"
+            );
+        }
+        if (batchTriggerAcquisitionFireAheadTimeWindowMillis < 0
+                || batchTriggerAcquisitionFireAheadTimeWindowMillis > 60_000L) {
+            throw new IllegalArgumentException(
+                "batchTriggerAcquisitionFireAheadTimeWindowMillis must be between 0 and 60000"
+            );
+        }
         if (clusterCheckinIntervalMillis < 1000) {
             throw new IllegalArgumentException(
                 "clusterCheckinIntervalMillis must be at least 1000"
@@ -85,6 +101,8 @@ public record QuartzSchedulerConfig(
             true,
             8,
             Thread.NORM_PRIORITY,
+            1,
+            0L,
             15_000L,
             60_000L,
             true,

easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java

@@ -154,7 +154,7 @@ public final class QuartzSchedulerFactory {
      * @param dataSourceName Quartz 内部 DataSource 名称
      * @return Quartz 属性
      */
-    private static Properties properties(
+    static Properties properties(
         QuartzSchedulerConfig config,
         String dataSourceName
     ) {
@@ -173,6 +173,14 @@ public final class QuartzSchedulerFactory {
             "org.quartz.threadPool.threadPriority",
             Integer.toString(config.threadPriority())
         );
+        properties.setProperty(
+            "org.quartz.scheduler.batchTriggerAcquisitionMaxCount",
+            Integer.toString(config.batchTriggerAcquisitionMaxCount())
+        );
+        properties.setProperty(
+            "org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow",
+            Long.toString(config.batchTriggerAcquisitionFireAheadTimeWindowMillis())
+        );
         properties.setProperty(
             "org.quartz.jobStore.class",
             "org.quartz.impl.jdbcjobstore.JobStoreTX"
@@ -182,6 +190,10 @@ public final class QuartzSchedulerFactory {
             config.driverDelegateClass()
         );
         properties.setProperty("org.quartz.jobStore.useProperties", "true");
+        properties.setProperty(
+            "org.quartz.jobStore.acquireTriggersWithinLock",
+            Boolean.toString(config.batchTriggerAcquisitionMaxCount() > 1)
+        );
         properties.setProperty("org.quartz.jobStore.dataSource", dataSourceName);
         properties.setProperty("org.quartz.jobStore.tablePrefix", config.tablePrefix());
         properties.setProperty(

easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java

@@ -9,6 +9,7 @@ import com.easyagents.scheduler.ScheduleException;
 import com.easyagents.scheduler.ScheduleFireContext;
 import com.easyagents.scheduler.ScheduleHandler;
 import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
 import org.h2.jdbcx.JdbcDataSource;
 import org.h2.tools.RunScript;
 import org.junit.Test;
@@ -20,9 +21,11 @@ import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.Statement;
 import java.time.Instant;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -218,6 +221,83 @@ public class QuartzJdbcIntegrationTest {
         }
     }
 
+    /**
+     * 验证延迟重试先落入 JDBC JobStore，关闭并重建节点后仍使用原 fire 上下文执行。
+     */
+    @Test
+    public void shouldRecoverPersistedRetryAfterSchedulerRestart() throws Exception {
+        JdbcDataSource dataSource = dataSource();
+        executeSchema(dataSource);
+        QuartzSchedulerConfig config = jdbcConfig();
+        CountDownLatch firstFailed = new CountDownLatch(1);
+        CountDownLatch retryCompleted = new CountDownLatch(1);
+        CopyOnWriteArrayList<ScheduleFireContext> contexts = new CopyOnWriteArrayList<>();
+        ScheduleHandler retryingHandler = new ScheduleHandler() {
+            @Override
+            public String code() {
+                return "persistent-retry-handler";
+            }
+
+            @Override
+            public void execute(ScheduleFireContext context) {
+                contexts.add(context);
+                if (contexts.size() == 1) {
+                    firstFailed.countDown();
+                    throw new ScheduleRefireException("temporary database failure",
+                        new IllegalStateException("unavailable"), 2,
+                        Duration.ofSeconds(2));
+                }
+                retryCompleted.countDown();
+            }
+        };
+        ScheduleId scheduleId = new ScheduleId("jdbc", "persistent-retry");
+        QuartzScheduleService first = QuartzSchedulerFactory.createJdbc(
+            dataSource, config, List.of(retryingHandler), List.of());
+        first.start();
+        first.create(new ScheduleDefinition(
+            scheduleId,
+            retryingHandler.code(),
+            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
+            MisfirePolicy.FIRE_ONCE_NOW,
+            ConcurrencyPolicy.DISALLOW,
+            true,
+            Map.of("snapshot", "original"),
+            "persistent retry"
+        ));
+        first.triggerNow(scheduleId, "persistent-invocation", Map.of());
+        assertTrue(firstFailed.await(5, TimeUnit.SECONDS));
+        awaitRetryTrigger(dataSource);
+        first.close();
+
+        QuartzScheduleService restarted = QuartzSchedulerFactory.createJdbc(
+            dataSource, config, List.of(retryingHandler), List.of());
+        try {
+            restarted.start();
+            assertTrue("persisted retry did not execute after restart",
+                retryCompleted.await(8, TimeUnit.SECONDS));
+            assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime());
+            assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime());
+            assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId());
+            assertEquals("persistent-invocation", contexts.get(1).invocationId());
+        } finally {
+            restarted.close();
+        }
+    }
+
+    private static void awaitRetryTrigger(JdbcDataSource dataSource) throws Exception {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        do {
+            try (Connection connection = dataSource.getConnection();
+                 Statement statement = connection.createStatement();
+                 ResultSet result = statement.executeQuery(
+                     "SELECT COUNT(*) FROM QRTZ_TRIGGERS WHERE TRIGGER_GROUP LIKE 'ea.scheduler.retry.%'")) {
+                if (result.next() && result.getInt(1) > 0) return;
+            }
+            Thread.sleep(25L);
+        } while (System.nanoTime() < deadline);
+        fail("persistent retry trigger was not stored");
+    }
+
     private static JdbcDataSource dataSource() {
         JdbcDataSource dataSource = new JdbcDataSource();
         dataSource.setURL("jdbc:h2:mem:scheduler-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
@@ -259,6 +339,8 @@ public class QuartzJdbcIntegrationTest {
             false,
             2,
             Thread.NORM_PRIORITY,
+            1,
+            0L,
             15_000L,
             1_000L,
             true,

easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java

@@ -0,0 +1,105 @@
+package com.easyagents.scheduler.quartz;
+
+import com.easyagents.scheduler.ConcurrencyPolicy;
+import com.easyagents.scheduler.MisfirePolicy;
+import com.easyagents.scheduler.OnceSchedulePlan;
+import com.easyagents.scheduler.ScheduleDefinition;
+import com.easyagents.scheduler.ScheduleHandler;
+import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
+import org.junit.Test;
+import org.quartz.JobDetail;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.Trigger;
+
+import java.lang.reflect.Proxy;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** {@link QuartzRuntime} 失败控制语义测试。 */
+public class QuartzRuntimeTest {
+
+    /** JobStore 无法保存延迟触发时不得把当前 fire 当作正常完成。 */
+    @Test
+    public void shouldRefireInPlaceWhenPersistentRetryCannotBeStored() throws Exception {
+        ScheduleDefinition definition = new ScheduleDefinition(
+            new ScheduleId("test", "retry-store-failure"),
+            "handler",
+            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
+            MisfirePolicy.FIRE_ONCE_NOW,
+            ConcurrencyPolicy.DISALLOW,
+            true,
+            Map.of(),
+            "retry store failure"
+        );
+        JobDetail job = QuartzScheduleMapper.toJobDetail(definition);
+        Scheduler scheduler = proxy(Scheduler.class, (proxy, method, arguments) -> {
+            if ("scheduleJob".equals(method.getName())
+                    && arguments != null && arguments.length == 1
+                    && arguments[0] instanceof Trigger) {
+                throw new SchedulerException("job store unavailable");
+            }
+            if ("isShutdown".equals(method.getName())) return false;
+            return defaultValue(method.getReturnType());
+        });
+        Date fireTime = new Date();
+        JobExecutionContext context = proxy(JobExecutionContext.class,
+            (proxy, method, arguments) -> switch (method.getName()) {
+                case "getMergedJobDataMap" -> job.getJobDataMap();
+                case "getJobDetail" -> job;
+                case "getScheduler" -> scheduler;
+                case "getFireTime", "getScheduledFireTime" -> fireTime;
+                case "getFireInstanceId" -> "fire-1";
+                case "isRecovering" -> false;
+                case "getRefireCount" -> 0;
+                default -> defaultValue(method.getReturnType());
+            });
+        QuartzRuntime runtime = new QuartzRuntime(List.of(new ScheduleHandler() {
+            @Override
+            public String code() {
+                return "handler";
+            }
+
+            @Override
+            public void execute(com.easyagents.scheduler.ScheduleFireContext context) {
+                throw new ScheduleRefireException("registration unavailable",
+                    new IllegalStateException("database unavailable"), 0,
+                    Duration.ofMillis(1));
+            }
+        }), List.of());
+
+        try {
+            runtime.execute(context);
+            fail("expected refire request");
+        } catch (JobExecutionException exception) {
+            assertTrue(exception.refireImmediately());
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T proxy(Class<T> type, java.lang.reflect.InvocationHandler handler) {
+        return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]{type}, handler);
+    }
+
+    private static Object defaultValue(Class<?> type) {
+        if (!type.isPrimitive()) return null;
+        if (type == boolean.class) return false;
+        if (type == byte.class) return (byte) 0;
+        if (type == short.class) return (short) 0;
+        if (type == int.class) return 0;
+        if (type == long.class) return 0L;
+        if (type == float.class) return 0F;
+        if (type == double.class) return 0D;
+        if (type == char.class) return '\0';
+        return null;
+    }
+}

easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java

@@ -10,20 +10,25 @@ import com.easyagents.scheduler.ScheduleException;
 import com.easyagents.scheduler.ScheduleFireContext;
 import com.easyagents.scheduler.ScheduleHandler;
 import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
 import com.easyagents.scheduler.ScheduleStatus;
 import org.junit.After;
 import org.junit.Test;
 import org.quartz.JobDetail;
 import org.quartz.Scheduler;
 import org.quartz.impl.StdSchedulerFactory;
+import org.quartz.impl.matchers.GroupMatcher;
 
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.util.Base64;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -112,6 +117,108 @@ public class QuartzScheduleServiceTest {
         assertEquals("value", captured.get().parameters().get("stable"));
     }
 
+    /** 验证可恢复失败会持久化延迟重试，并保留原始 fire 上下文。 */
+    @Test
+    public void shouldRefireRetryableHandlerFailure() throws Exception {
+        AtomicInteger attempts = new AtomicInteger();
+        CountDownLatch completed = new CountDownLatch(1);
+        CopyOnWriteArrayList<ScheduleFireContext> contexts = new CopyOnWriteArrayList<>();
+        service = newRamService(context -> {
+            contexts.add(context);
+            if (attempts.incrementAndGet() == 1) {
+                throw new ScheduleRefireException("temporary registration failure",
+                    new IllegalStateException("database unavailable"), 2,
+                    Duration.ofMillis(100));
+            }
+            completed.countDown();
+        });
+        ScheduleDefinition definition = onceDefinition(
+            "retryable-immediate",
+            MisfirePolicy.FIRE_ONCE_NOW,
+            ConcurrencyPolicy.DISALLOW,
+            Instant.parse("2099-01-01T00:00:00Z")
+        );
+        service.create(definition);
+
+        service.triggerNow(definition.id(), "retryable-invocation", Map.of());
+
+        assertTrue("retryable handler was not refired", completed.await(5, TimeUnit.SECONDS));
+        assertEquals(2, attempts.get());
+        assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime());
+        assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime());
+        assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId());
+        assertEquals("retryable-invocation", contexts.get(1).invocationId());
+    }
+
+    /** 持久化延迟重试不得占用当前 Quartz worker。 */
+    @Test
+    public void shouldReleaseWorkerWhilePersistentRetryIsDelayed() throws Exception {
+        CountDownLatch retryScheduled = new CountDownLatch(1);
+        CountDownLatch healthyCompleted = new CountDownLatch(1);
+        CountDownLatch retryCompleted = new CountDownLatch(1);
+        AtomicInteger retryAttempts = new AtomicInteger();
+        service = newRamService(context -> {
+            if (context.scheduleId().name().equals("delayed-retry")) {
+                if (retryAttempts.incrementAndGet() == 1) {
+                    retryScheduled.countDown();
+                    throw new ScheduleRefireException("temporary registration failure",
+                        new IllegalStateException("database unavailable"), 1,
+                        Duration.ofSeconds(1));
+                }
+                retryCompleted.countDown();
+            } else {
+                healthyCompleted.countDown();
+            }
+        }, 30_000L, 1);
+        ScheduleDefinition retry = onceDefinition(
+            "delayed-retry", MisfirePolicy.FIRE_ONCE_NOW,
+            ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z"));
+        ScheduleDefinition healthy = onceDefinition(
+            "healthy-during-retry", MisfirePolicy.FIRE_ONCE_NOW,
+            ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z"));
+        service.create(retry);
+        service.create(healthy);
+
+        service.triggerNow(retry.id(), "retry-1", Map.of());
+        assertTrue(retryScheduled.await(5, TimeUnit.SECONDS));
+        service.triggerNow(healthy.id(), "healthy-1", Map.of());
+
+        assertTrue("single Quartz worker remained occupied by delayed retry",
+            healthyCompleted.await(750, TimeUnit.MILLISECONDS));
+        assertTrue(retryCompleted.await(5, TimeUnit.SECONDS));
+    }
+
+    /** 已持久化 fire 的重试不得因 replace 而切换到新 Handler。 */
+    @Test
+    public void shouldKeepOriginalHandlerWhenScheduleIsReplacedDuringRetry() throws Exception {
+        AtomicInteger oldAttempts = new AtomicInteger();
+        AtomicInteger newAttempts = new AtomicInteger();
+        CountDownLatch oldRetryCompleted = new CountDownLatch(1);
+        service = newRamServiceWithHandlers(List.of(
+            handler("old-handler", context -> {
+                if (oldAttempts.incrementAndGet() == 1) {
+                    throw new ScheduleRefireException("temporary registration failure",
+                        new IllegalStateException("database unavailable"), 1,
+                        Duration.ofMillis(400));
+                }
+                oldRetryCompleted.countDown();
+            }),
+            handler("new-handler", context -> newAttempts.incrementAndGet())
+        ), 30_000L, 2);
+        ScheduleId id = new ScheduleId("test", "replace-during-retry");
+        ScheduleDefinition original = definition(id, "old-handler");
+        service.create(original);
+        service.triggerNow(id, "replace-retry-1", Map.of());
+        awaitPersistentRetryTrigger();
+
+        service.replace(definition(id, "new-handler"));
+
+        assertTrue("old handler retry did not complete",
+            oldRetryCompleted.await(5, TimeUnit.SECONDS));
+        assertEquals(2, oldAttempts.get());
+        assertEquals(0, newAttempts.get());
+    }
+
     /**
      * 验证立即触发会在返回回执前校验基础参数与覆盖参数的合并结果。
      */
@@ -385,6 +492,23 @@ public class QuartzScheduleServiceTest {
     private QuartzScheduleService newRamService(
         Consumer<ScheduleFireContext> handler,
         long shutdownWaitTimeoutMillis
+    ) throws Exception {
+        return newRamService(handler, shutdownWaitTimeoutMillis, 2);
+    }
+
+    private QuartzScheduleService newRamService(
+        Consumer<ScheduleFireContext> handler,
+        long shutdownWaitTimeoutMillis,
+        int threadCount
+    ) throws Exception {
+        return newRamServiceWithHandlers(List.of(handler("handler", handler)),
+            shutdownWaitTimeoutMillis, threadCount);
+    }
+
+    private QuartzScheduleService newRamServiceWithHandlers(
+        List<ScheduleHandler> handlers,
+        long shutdownWaitTimeoutMillis,
+        int threadCount
     ) throws Exception {
         Properties properties = new Properties();
         properties.setProperty(
@@ -394,7 +518,7 @@ public class QuartzScheduleServiceTest {
         properties.setProperty("org.quartz.scheduler.instanceId", "NON_CLUSTERED");
         properties.setProperty("org.quartz.scheduler.interruptJobsOnShutdown", "true");
         properties.setProperty("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
-        properties.setProperty("org.quartz.threadPool.threadCount", "2");
+        properties.setProperty("org.quartz.threadPool.threadCount", Integer.toString(threadCount));
         properties.setProperty("org.quartz.jobStore.class", "org.quartz.simpl.RAMJobStore");
         properties.setProperty("org.quartz.jobStore.misfireThreshold", "100");
         Scheduler scheduler = new StdSchedulerFactory(properties).getScheduler();
@@ -402,17 +526,7 @@ public class QuartzScheduleServiceTest {
             scheduler,
             true,
             shutdownWaitTimeoutMillis,
-            java.util.List.of(new ScheduleHandler() {
+            handlers,
-                @Override
-                public String code() {
-                    return "handler";
-                }
-
-                @Override
-                public void execute(ScheduleFireContext context) {
-                    handler.accept(context);
-                }
-            }),
             java.util.List.of()
         );
         result.start();
@@ -420,6 +534,50 @@ public class QuartzScheduleServiceTest {
         return result;
     }
 
+    private void awaitPersistentRetryTrigger() throws Exception {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (System.nanoTime() < deadline) {
+            boolean found = service.quartzScheduler()
+                .getTriggerKeys(GroupMatcher.anyTriggerGroup())
+                .stream()
+                .anyMatch(key -> key.getGroup().startsWith(
+                    QuartzScheduleMapper.GROUP_PREFIX + "retry."));
+            if (found) return;
+            Thread.sleep(10L);
+        }
+        fail("persistent retry trigger was not created");
+    }
+
+    private static ScheduleHandler handler(
+        String code,
+        Consumer<ScheduleFireContext> consumer
+    ) {
+        return new ScheduleHandler() {
+            @Override
+            public String code() {
+                return code;
+            }
+
+            @Override
+            public void execute(ScheduleFireContext context) {
+                consumer.accept(context);
+            }
+        };
+    }
+
+    private static ScheduleDefinition definition(ScheduleId id, String handlerCode) {
+        return new ScheduleDefinition(
+            id,
+            handlerCode,
+            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
+            MisfirePolicy.FIRE_ONCE_NOW,
+            ConcurrencyPolicy.DISALLOW,
+            true,
+            Map.of(),
+            id.name()
+        );
+    }
+
     private static ScheduleDefinition cronDefinition(String handlerCode, String expression) {
         return new ScheduleDefinition(
             new ScheduleId("test", "lifecycle"),

easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java

@@ -0,0 +1,40 @@
+package com.easyagents.scheduler.quartz;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+/** {@link QuartzSchedulerFactory} 原生属性映射测试。 */
+public class QuartzSchedulerFactoryTest {
+
+    @Test
+    public void batchAcquisitionMustRunWithinJobStoreLock() {
+        QuartzSchedulerConfig config = new QuartzSchedulerConfig(
+            "batch-scheduler",
+            "NON_CLUSTERED",
+            "QRTZ_",
+            QuartzSchedulerConfig.STANDARD_JDBC_DELEGATE,
+            false,
+            8,
+            Thread.NORM_PRIORITY,
+            8,
+            1_000L,
+            15_000L,
+            60_000L,
+            true,
+            30_000L,
+            false
+        );
+
+        Properties properties = QuartzSchedulerFactory.properties(config, "testDs");
+
+        assertEquals("8", properties.getProperty(
+            "org.quartz.scheduler.batchTriggerAcquisitionMaxCount"));
+        assertEquals("1000", properties.getProperty(
+            "org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow"));
+        assertEquals("true", properties.getProperty(
+            "org.quartz.jobStore.acquireTriggersWithinLock"));
+    }
+}

easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml

@@ -53,6 +53,11 @@
             <artifactId>h2</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-jdbc</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java

@@ -18,6 +18,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.sql.init.dependency.DependsOnDatabaseInitialization;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -61,6 +62,7 @@ public class EasyAgentsSchedulerAutoConfiguration {
      */
     @Bean(name = SCHEDULER_BEAN_NAME, initMethod = "start", destroyMethod = "close")
     @ConditionalOnMissingBean(ScheduleService.class)
+    @DependsOnDatabaseInitialization
     public QuartzScheduleService easyAgentsQuartzScheduleService(
         EasyAgentsSchedulerProperties properties,
         ListableBeanFactory beanFactory,
@@ -83,6 +85,8 @@ public class EasyAgentsSchedulerAutoConfiguration {
             quartz.isClustered(),
             quartz.getThreadCount(),
             quartz.getThreadPriority(),
+            quartz.getBatchTriggerAcquisitionMaxCount(),
+            quartz.getBatchTriggerAcquisitionFireAheadTimeWindowMillis(),
             quartz.getClusterCheckinIntervalMillis(),
             quartz.getMisfireThresholdMillis(),
             quartz.isWaitForJobsToCompleteOnShutdown(),

easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java

@@ -104,6 +104,12 @@ public class EasyAgentsSchedulerProperties {
         /** Quartz Worker 线程优先级。 */
         private int threadPriority = Thread.NORM_PRIORITY;
 
+        /** 单次批量获取 Trigger 的最大数量。 */
+        private int batchTriggerAcquisitionMaxCount = 1;
+
+        /** 可提前纳入批量的时间窗口，单位毫秒。 */
+        private long batchTriggerAcquisitionFireAheadTimeWindowMillis;
+
         /** 集群心跳间隔，单位毫秒。 */
         private long clusterCheckinIntervalMillis = 15_000L;
 
@@ -251,6 +257,45 @@ public class EasyAgentsSchedulerProperties {
             this.threadPriority = threadPriority;
         }
 
+        /**
+         * 返回单次批量获取 Trigger 的最大数量。
+         *
+         * @return 批量上限
+         */
+        public int getBatchTriggerAcquisitionMaxCount() {
+            return batchTriggerAcquisitionMaxCount;
+        }
+
+        /**
+         * 设置单次批量获取 Trigger 的最大数量。
+         *
+         * @param batchTriggerAcquisitionMaxCount 批量上限
+         */
+        public void setBatchTriggerAcquisitionMaxCount(
+                int batchTriggerAcquisitionMaxCount) {
+            this.batchTriggerAcquisitionMaxCount = batchTriggerAcquisitionMaxCount;
+        }
+
+        /**
+         * 返回可提前纳入批量的时间窗口。
+         *
+         * @return 毫秒窗口
+         */
+        public long getBatchTriggerAcquisitionFireAheadTimeWindowMillis() {
+            return batchTriggerAcquisitionFireAheadTimeWindowMillis;
+        }
+
+        /**
+         * 设置可提前纳入批量的时间窗口。
+         *
+         * @param batchTriggerAcquisitionFireAheadTimeWindowMillis 毫秒窗口
+         */
+        public void setBatchTriggerAcquisitionFireAheadTimeWindowMillis(
+                long batchTriggerAcquisitionFireAheadTimeWindowMillis) {
+            this.batchTriggerAcquisitionFireAheadTimeWindowMillis =
+                batchTriggerAcquisitionFireAheadTimeWindowMillis;
+        }
+
         /**
          * 返回集群心跳间隔。
          *

easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java

@@ -114,6 +114,47 @@ public class EasyAgentsSchedulerAutoConfigurationTest {
         }
     }
 
+    /**
+     * 验证调度器等待 Spring Boot 数据库脚本初始化完成后再启动。
+     */
+    @Test
+    public void shouldWaitForDatabaseInitializationBeforeStartingScheduler() throws Exception {
+        JdbcDataSource dataSource = dataSource();
+        Map<String, Object> properties = enabledProperties();
+        properties.put("easy-agents.scheduler.data-source-bean-name", "schedulerDataSource");
+        properties.put("spring.sql.init.mode", "always");
+        properties.put(
+            "spring.sql.init.schema-locations",
+            "classpath:quartz-schema/h2-2.5.2.sql"
+        );
+
+        SpringApplication application = new SpringApplication(AutoDiscoveryApplication.class);
+        application.setWebApplicationType(WebApplicationType.NONE);
+        application.setDefaultProperties(properties);
+        application.addInitializers(applicationContext -> {
+            GenericApplicationContext genericContext =
+                (GenericApplicationContext) applicationContext;
+            genericContext.registerBean(
+                "schedulerDataSource",
+                DataSource.class,
+                () -> dataSource
+            );
+        });
+
+        try (ConfigurableApplicationContext context = application.run()) {
+            assertNotNull(context.getBean(ScheduleService.class));
+            try (
+                Connection connection = dataSource.getConnection();
+                Statement statement = connection.createStatement();
+                ResultSet resultSet = statement.executeQuery(
+                    "SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE"
+                )
+            ) {
+                assertTrue(resultSet.next());
+            }
+        }
+    }
+
     /**
      * 验证多个 DataSource 未明确选择时启动失败并提供可操作信息。
      */
@@ -162,6 +203,13 @@ public class EasyAgentsSchedulerAutoConfigurationTest {
         properties.put("easy-agents.scheduler.quartz.instance-id", "NON_CLUSTERED");
         properties.put("easy-agents.scheduler.quartz.clustered", "false");
         properties.put("easy-agents.scheduler.quartz.thread-count", "2");
+        properties.put(
+            "easy-agents.scheduler.quartz.batch-trigger-acquisition-max-count", "2"
+        );
+        properties.put(
+            "easy-agents.scheduler.quartz.batch-trigger-acquisition-fire-ahead-time-window-millis",
+            "500"
+        );
         properties.put("easy-agents.scheduler.quartz.misfire-threshold-millis", "1000");
         return properties;
     }