{@link CronJobService} runs in every node of a multi-instance + * deployment; without coordination, every node fires every {@code CronTrigger} + * tick, multiplying invocations and downstream side effects (channel + * messages, LLM calls, approval rows). ShedLock's JDBC mode reuses the + * existing application DataSource so we don't pull in Redis just for this + * one purpose. + * + *
Schema lives in {@code db/migration/{h2,mysql}/V74__shedlock_table.sql}. + * + *
Default {@code lockAtMostFor=PT30M} on the {@link EnableSchedulerLock} + * annotation is the safety net for a node that dies mid-execution — after + * 30 min any other node can take the lock. Per-call {@code @SchedulerLock} + * annotations may shorten this for predictable workloads. + * + *
Single-node deployments (desktop / single docker container) are + * unaffected: the lock is acquired on the same node trivially. + */ +@Slf4j +@Configuration +@EnableSchedulerLock(defaultLockAtMostFor = "PT30M") +public class ShedLockConfig { + + @Bean + public LockProvider lockProvider(DataSource dataSource) { + log.info("[ShedLock] Initializing JDBC LockProvider for cron scheduling"); + return new JdbcTemplateLockProvider( + JdbcTemplateLockProvider.Configuration.builder() + .withJdbcTemplate(new JdbcTemplate(dataSource)) + .withTableName("shedlock") + .usingDbTime() // server-side NOW() — avoids node clock drift + .build() + ); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java index 4b5fa171..7d5efb41 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java @@ -3,6 +3,7 @@ package vip.mate.cron.delivery; import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import net.javacrumbs.shedlock.spring.annotation.SchedulerLock; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import vip.mate.dashboard.model.CronJobRunEntity; @@ -39,7 +40,17 @@ public class CronRunStaleCleanup { private static final Duration DELIVERY_STALE = Duration.ofMinutes(15); private static final Duration RUN_STALE = Duration.ofMinutes(30); + /** + * RFC-03 Lane G2: in a multi-instance deployment, the sweep is purely + * idempotent (UPDATE with predicates) so duplicates would be harmless, + * but locking still saves N-1 nodes the DB roundtrips and keeps the + * dashboard counters honest. {@code lockAtMostFor} comfortably exceeds + * the worst-case sweep latency we have seen (≪1s). + */ @Scheduled(fixedDelay = 5 * 60 * 1000L, initialDelay = 60 * 1000L) + @SchedulerLock(name = "cronRunStaleCleanup", + lockAtMostFor = "PT2M", + lockAtLeastFor = "PT30S") public void sweep() { LocalDateTime now = LocalDateTime.now(); diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java index 97c681f7..fa121173 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java @@ -5,6 +5,9 @@ import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import net.javacrumbs.shedlock.core.LockConfiguration; +import net.javacrumbs.shedlock.core.LockProvider; +import net.javacrumbs.shedlock.core.SimpleLock; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.core.annotation.Order; @@ -22,12 +25,15 @@ import vip.mate.cron.model.CronJobEntity; import vip.mate.cron.repository.CronJobMapper; import vip.mate.exception.MateClawException; +import java.time.Duration; +import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; import java.time.ZonedDateTime; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -50,6 +56,13 @@ public class CronJobService implements ApplicationRunner { private final CronJobMapper cronJobMapper; private final AgentMapper agentMapper; private final ChannelMapper channelMapper; + /** + * RFC-03 Lane G2: distributed lock for fire-time execution. ShedLock's + * JDBC provider is configured in {@link vip.mate.cron.config.ShedLockConfig} + * — see also {@link #LOCK_AT_MOST_FOR} / {@link #LOCK_AT_LEAST_FOR} + * tuning notes on {@link #register}. + */ + private final LockProvider lockProvider; /** * RFC-063r §2.7.1: cron-tick execution moved to {@link CronJobRunner} * (separate bean) so the three-segment transactional model in @@ -94,6 +107,23 @@ public class CronJobService implements ApplicationRunner { private static final int MAX_CONCURRENT_CRON_RUNS = 8; private final Semaphore cronConcurrencyLimiter = new Semaphore(MAX_CONCURRENT_CRON_RUNS); + /** + * RFC-03 Lane G2: ShedLock duration tuning. + * + *
{@code lockAtMostFor} is the safety net for a node that crashes + * mid-execution — after this window, any other node may take over the + * job's next tick. 30 minutes covers all observed cron run times + * (longest LLM-driven jobs in production sit around p99 ≈ 8 min). + * + *
{@code lockAtLeastFor} prevents thundering-herd when a fast job + * (e.g. trivial SQL query that completes in <1s) finishes before its + * own next tick — without it, the same node could fire twice per tick + * if its clock is slightly ahead. 30s gives every other node a chance + * to see the lock as held even for instant-finishing jobs. + */ + private static final Duration LOCK_AT_MOST_FOR = Duration.ofMinutes(30); + private static final Duration LOCK_AT_LEAST_FOR = Duration.ofSeconds(30); + // ==================== 初始化与销毁 ==================== /** @@ -403,8 +433,12 @@ public class CronJobService implements ApplicationRunner { // cronExecutor — the LLM call must NOT run on a scheduler // worker (4 concurrent long crons would otherwise saturate the // pool and the 5th would miss its tick). - ScheduledFuture future = scheduler.schedule(() -> - cronExecutor.submit(() -> runWithBackpressure(job, "scheduled")), trigger); + // + // RFC-03 Lane G2: tickWithDistributedLock wraps the offload in + // a ShedLock acquire/release so a multi-instance deployment + // fires each tick exactly once. See javadoc on that method. + ScheduledFuture future = scheduler.schedule( + () -> tickWithDistributedLock(job), trigger); scheduledTasks.put(job.getId(), future); log.info("[CronJob] Registered job {} ({}) ws={}, cron={}, tz={}", job.getId(), job.getName(), job.getWorkspaceId(), @@ -421,6 +455,48 @@ public class CronJobService implements ApplicationRunner { } } + /** + * RFC-03 Lane G2 — distributed-lock-guarded tick fire. + * + *
Called on the scheduler thread when {@link CronTrigger} fires. + * Tries to acquire a ShedLock entry keyed by {@code "cron-job-{jobId}"}; + * if another node holds it, returns immediately (silent skip — siblings + * always see this for every tick, which is by design). On success, + * passes lock ownership into the virtual-thread executor so the work + * proceeds on {@link #cronExecutor} and the lock releases only after + * {@code runWithBackpressure} completes. {@link #LOCK_AT_MOST_FOR} is + * the safety net for a node that crashes mid-execution. + * + *
Package-private so unit tests can drive lock-acquisition outcomes
+ * without booting the full Spring context.
+ */
+ void tickWithDistributedLock(CronJobEntity job) {
+ Long jobId = job.getId();
+ Optional