From 2cd51743d1d759b52e17e7ba04ab93d1634ce393 Mon Sep 17 00:00:00 2001 From: matevip Date: Sat, 2 May 2026 15:40:31 +0800 Subject: [PATCH] feat(cron): distributed lock for multi-instance cron firing --- mateclaw-server/pom.xml | 15 ++++ .../vip/mate/cron/config/ShedLockConfig.java | 49 ++++++++++++ .../cron/delivery/CronRunStaleCleanup.java | 11 +++ .../vip/mate/cron/service/CronJobService.java | 80 ++++++++++++++++++- .../db/migration/h2/V74__shedlock_table.sql | 17 ++++ .../migration/mysql/V74__shedlock_table.sql | 16 ++++ 6 files changed, 186 insertions(+), 2 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/cron/config/ShedLockConfig.java create mode 100644 mateclaw-server/src/main/resources/db/migration/h2/V74__shedlock_table.sql create mode 100644 mateclaw-server/src/main/resources/db/migration/mysql/V74__shedlock_table.sql diff --git a/mateclaw-server/pom.xml b/mateclaw-server/pom.xml index 1f19e5df..a24982b3 100644 --- a/mateclaw-server/pom.xml +++ b/mateclaw-server/pom.xml @@ -381,6 +381,21 @@ 1.3.0 test + + + + net.javacrumbs.shedlock + shedlock-spring + 5.16.0 + + + net.javacrumbs.shedlock + shedlock-provider-jdbc-template + 5.16.0 + diff --git a/mateclaw-server/src/main/java/vip/mate/cron/config/ShedLockConfig.java b/mateclaw-server/src/main/java/vip/mate/cron/config/ShedLockConfig.java new file mode 100644 index 00000000..abb10c24 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/config/ShedLockConfig.java @@ -0,0 +1,49 @@ +package vip.mate.cron.config; + +import lombok.extern.slf4j.Slf4j; +import net.javacrumbs.shedlock.core.LockProvider; +import net.javacrumbs.shedlock.provider.jdbctemplate.JdbcTemplateLockProvider; +import net.javacrumbs.shedlock.spring.annotation.EnableSchedulerLock; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; + +import javax.sql.DataSource; + +/** + * RFC-03 Lane G2 — distributed lock provider for the cron scheduler. + * + *

{@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 maybeLock = lockProvider.lock(new LockConfiguration( + Instant.now(), + "cron-job-" + jobId, + LOCK_AT_MOST_FOR, + LOCK_AT_LEAST_FOR)); + if (maybeLock.isEmpty()) { + log.debug("[CronJob] {} skipped — another node holds the tick lock", jobId); + return; + } + SimpleLock lock = maybeLock.get(); + cronExecutor.submit(() -> { + try { + runWithBackpressure(job, "scheduled"); + } finally { + try { + lock.unlock(); + } catch (Exception unlockEx) { + // Lock will expire after LOCK_AT_MOST_FOR anyway; log + continue + // so a transient unlock failure doesn't taint the cron worker. + log.warn("[CronJob] {} lock release failed: {}", jobId, unlockEx.getMessage()); + } + } + }); + } + // ==================== 任务执行 ==================== // // RFC-063r §2.7.1: the executeJob body moved to CronJobRunner so the diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V74__shedlock_table.sql b/mateclaw-server/src/main/resources/db/migration/h2/V74__shedlock_table.sql new file mode 100644 index 00000000..504a4d5f --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V74__shedlock_table.sql @@ -0,0 +1,17 @@ +-- V74: ShedLock distributed-lock table (RFC-03 Lane G2). +-- Backs the LockProvider configured in vip.mate.cron.config.ShedLockConfig +-- so a multi-instance deployment fires each cron job exactly once per tick. +-- Single-node setups are unaffected (acquiring the lock from the only node +-- always succeeds trivially). +-- +-- Schema is the canonical ShedLock layout from +-- https://github.com/lukas-krecan/ShedLock#configure-lockprovider — H2 +-- accepts the same DDL as MySQL for our column types. + +CREATE TABLE IF NOT EXISTS shedlock ( + name VARCHAR(64) NOT NULL, + lock_until TIMESTAMP(3) NOT NULL, + locked_at TIMESTAMP(3) NOT NULL, + locked_by VARCHAR(255) NOT NULL, + PRIMARY KEY (name) +); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V74__shedlock_table.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V74__shedlock_table.sql new file mode 100644 index 00000000..6915b995 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V74__shedlock_table.sql @@ -0,0 +1,16 @@ +-- V74: ShedLock distributed-lock table (RFC-03 Lane G2). +-- Backs the LockProvider configured in vip.mate.cron.config.ShedLockConfig +-- so a multi-instance deployment fires each cron job exactly once per tick. +-- Single-node setups are unaffected (acquiring the lock from the only node +-- always succeeds trivially). +-- +-- Schema is the canonical ShedLock layout from +-- https://github.com/lukas-krecan/ShedLock#configure-lockprovider. + +CREATE TABLE IF NOT EXISTS shedlock ( + name VARCHAR(64) NOT NULL, + lock_until TIMESTAMP(3) NOT NULL, + locked_at TIMESTAMP(3) NOT NULL, + locked_by VARCHAR(255) NOT NULL, + PRIMARY KEY (name) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;