opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(cron): distributed lock for multi-instance cron firing

Browse Source
This commit is contained in:
matevip 2026-05-02 15:40:31 +08:00
parent 92bd8e9b6e
commit 2cd51743d1
6 changed files with 186 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
mateclaw-server/pom.xml
View File

@ -381,6 +381,21 @@
<version>1.3.0</version>
<scope>test</scope>
</dependency>
<!-- RFC-03 Lane G2: distributed lock for the cron scheduler so a
multi-instance deployment doesn't fire the same job N times.
JDBC mode reuses the existing DataSource — no Redis dependency
on the desktop / single-node footprint. -->
<dependency>
<groupId>net.javacrumbs.shedlock</groupId>
<artifactId>shedlock-spring</artifactId>
<version>5.16.0</version>
</dependency>
<dependency>
<groupId>net.javacrumbs.shedlock</groupId>
<artifactId>shedlock-provider-jdbc-template</artifactId>
<version>5.16.0</version>
</dependency>
</dependencies>
<build>

49
mateclaw-server/src/main/java/vip/mate/cron/config/ShedLockConfig.java Normal file
View File

@ -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.
*
* <p>{@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.
*
* <p>Schema lives in {@code db/migration/{h2,mysql}/V74__shedlock_table.sql}.
*
* <p>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.
*
* <p>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()
);
}
}

11
mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java
View File

@ -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();

80
mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java
View File

@ -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.
*
* <p>{@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).
*
* <p>{@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.
*
* <p>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.
*
* <p>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<SimpleLock> 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

17
mateclaw-server/src/main/resources/db/migration/h2/V74__shedlock_table.sql Normal file
View File

@ -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)
);

16
mateclaw-server/src/main/resources/db/migration/mysql/V74__shedlock_table.sql Normal file
View File

@ -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;