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 62276e9a..a1acf390 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 @@ -8,6 +8,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.core.annotation.Order; +import org.springframework.dao.DuplicateKeyException; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.CronExpression; import org.springframework.scheduling.support.CronTrigger; @@ -31,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.Semaphore; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; @@ -79,6 +81,19 @@ public class CronJobService implements ApplicationRunner { private final ExecutorService cronExecutor = Executors.newThreadPerTaskExecutor( Thread.ofVirtual().name("cron-execute-", 0).factory()); + /** + * Issue #50: cap concurrent cron run executions so that hundreds of jobs + * firing at the same minute boundary cannot exhaust the JDBC pool. Each + * run holds 3-4 connections in sequence (three REQUIRES_NEW segments in + * {@link CronJobLifecycleService} plus {@link #updateRunTimes}); without + * a limit, virtual threads launch unboundedly and starve the channel + * monitor / web traffic. Tuned alongside Hikari maximum-pool-size — keep + * this value well below the pool size so non-cron paths still get + * connections. + */ + private static final int MAX_CONCURRENT_CRON_RUNS = 8; + private final Semaphore cronConcurrencyLimiter = new Semaphore(MAX_CONCURRENT_CRON_RUNS); + // ==================== 初始化与销毁 ==================== /** @@ -181,6 +196,18 @@ public class CronJobService implements ApplicationRunner { // toSpringCron 校验表达式合法性，结果复用于后续 calcNextRunTime 和 register String springCron = toSpringCron(dto.getCronExpression()); + // Issue #50: dedup by (workspace_id, agent_id, name). LLM-driven + // creators (CronJobTool) call this on every retry; without this guard + // a single instruction can produce N identical rows that all fire on + // the same tick. App-level check covers the common case; the unique + // index added in V67 protects against races (handled below). + CronJobEntity duplicate = findActiveDuplicate(workspaceId, dto.getAgentId(), dto.getName()); + if (duplicate != null) { + log.info("[CronJob] create dedup hit: ws={} agent={} name={} → returning existing id={}", + workspaceId, dto.getAgentId(), dto.getName(), duplicate.getId()); + return getById(duplicate.getId(), workspaceId); + } + CronJobEntity entity = dto.toEntity(); // RFC-083: workspace stamped server-side from X-Workspace-Id; never // trust a client-supplied value (DTO.toEntity intentionally drops it). @@ -190,7 +217,18 @@ public class CronJobService implements ApplicationRunner { if (entity.getEnabled() == null) entity.setEnabled(true); entity.setNextRunTime(calcNextRunTime(springCron, entity.getTimezone())); - cronJobMapper.insert(entity); + try { + cronJobMapper.insert(entity); + } catch (DuplicateKeyException e) { + // Race: another concurrent create won. Re-fetch and return that one. + CronJobEntity raced = findActiveDuplicate(workspaceId, dto.getAgentId(), dto.getName()); + if (raced != null) { + log.info("[CronJob] create race resolved: ws={} agent={} name={} → existing id={}", + workspaceId, dto.getAgentId(), dto.getName(), raced.getId()); + return getById(raced.getId(), workspaceId); + } + throw e; + } if (Boolean.TRUE.equals(entity.getEnabled())) { // register() 内部会再次调用 toSpringCron，但表达式已校验过，不会抛异常 @@ -200,6 +238,22 @@ public class CronJobService implements ApplicationRunner { return getById(entity.getId(), workspaceId); } + /** + * Issue #50: lookup existing active row for the dedup natural key. + * No {@code @TableLogic} on this entity — {@code deleted=0} must be + * filtered explicitly. + */ + private CronJobEntity findActiveDuplicate(Long workspaceId, Long agentId, String name) { + if (workspaceId == null || agentId == null || name == null) return null; + return cronJobMapper.selectOne( + new LambdaQueryWrapper() + .eq(CronJobEntity::getWorkspaceId, workspaceId) + .eq(CronJobEntity::getAgentId, agentId) + .eq(CronJobEntity::getName, name) + .eq(CronJobEntity::getDeleted, 0) + .last("LIMIT 1")); + } + public CronJobDTO update(Long id, CronJobDTO dto, Long workspaceId) { // RFC-083: scoped lookup — cross-workspace updates 404 the same as // deleted rows. @@ -290,13 +344,7 @@ public class CronJobService implements ApplicationRunner { // CronJobLifecycleService work as advertised. "manual" trigger type // distinguishes this from scheduler-driven runs in mate_cron_job_run. // Run on the virtual-thread cronExecutor — never block the scheduler. - cronExecutor.submit(() -> { - try { - cronJobRunner.executeJob(entity, "manual"); - } finally { - updateRunTimes(entity.getId(), entity.getCronExpression(), entity.getTimezone()); - } - }); + cronExecutor.submit(() -> runWithBackpressure(entity, "manual")); } // ==================== 调度器管理 ==================== @@ -314,13 +362,7 @@ public class CronJobService implements ApplicationRunner { // worker (4 concurrent long crons would otherwise saturate the // pool and the 5th would miss its tick). ScheduledFuture future = scheduler.schedule(() -> - cronExecutor.submit(() -> { - try { - cronJobRunner.executeJob(job, "scheduled"); - } finally { - updateRunTimes(job.getId(), job.getCronExpression(), job.getTimezone()); - } - }), trigger); + cronExecutor.submit(() -> runWithBackpressure(job, "scheduled")), trigger); scheduledTasks.put(job.getId(), future); log.info("[CronJob] Registered job {} ({}) ws={}, cron={}, tz={}", job.getId(), job.getName(), job.getWorkspaceId(), @@ -351,6 +393,37 @@ public class CronJobService implements ApplicationRunner { // see those methods above. The wrap below ensures next-run rolls forward // regardless of run outcome. + /** + * Issue #50: gate every run on {@link #cronConcurrencyLimiter} so that a + * minute-boundary stampede of N enabled jobs cannot fan out into N + * simultaneous JDBC connection acquisitions. Excess runs queue on the + * virtual thread (cheap) instead of competing for the pool. + * + *

The {@code updateRunTimes} write is intentionally inside the + * permit's hold so the next-run pointer advances under the same + * backpressure budget — otherwise a tail of bookkeeping writes could + * still pile up after the executor "finishes". + */ + private void runWithBackpressure(CronJobEntity job, String triggerType) { + try { + cronConcurrencyLimiter.acquire(); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + log.warn("[CronJob] Interrupted while waiting to run job {} ({})", + job.getId(), triggerType); + return; + } + try { + cronJobRunner.executeJob(job, triggerType); + } finally { + try { + updateRunTimes(job.getId(), job.getCronExpression(), job.getTimezone()); + } finally { + cronConcurrencyLimiter.release(); + } + } + } + /** * 合并更新 lastRunTime 和 nextRunTime，单次 DB 写入替代原来的 4 次 selectById + updateById */ diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java index cff51f96..cf3ca8f7 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java @@ -40,7 +40,10 @@ public class CronJobTool { + "DO NOT use this for plain reminders where the user already wrote the exact text they want delivered — " + "use create_reminder instead, otherwise the LLM will rephrase or echo the reminder. " + "Use 5-field cron expressions: minute hour day month weekday. " - + "Examples: '0 9 * * *' = daily at 9am, '0 9 * * 1-5' = weekdays at 9am, '*/30 * * * *' = every 30 minutes.") + + "Examples: '0 9 * * *' = daily at 9am, '0 9 * * 1-5' = weekdays at 9am, '*/30 * * * *' = every 30 minutes. " + + "If a task with the same name already exists for this agent, the existing one is returned " + + "(deduplicated=true in the response) — do NOT call this tool again with a different name " + + "just to retry; check list_cron_jobs first if unsure.") public String create_cron_job( @ToolParam(description = "Task name, e.g. 'Daily AI News Summary'") String name, @ToolParam(description = "5-field cron expression: minute hour day month weekday") String cronExpression, @@ -114,7 +117,9 @@ public class CronJobTool { + "reminder text 'It's time to leave for the meeting'). " + "DO NOT use this if the message requires the agent to compute or look something up — use create_cron_job for that. " + "Use 5-field cron expressions: minute hour day month weekday. " - + "Examples: '0 15 * * *' = every day at 3pm, '0 9 * * 1' = every Monday at 9am.") + + "Examples: '0 15 * * *' = every day at 3pm, '0 9 * * 1' = every Monday at 9am. " + + "If a reminder with the same name already exists for this agent, the existing one is returned — " + + "do NOT retry with the same name to force a new row.") public String create_reminder( @ToolParam(description = "Reminder name, e.g. 'Meeting at Room 6'") String name, @ToolParam(description = "5-field cron expression: minute hour day month weekday") String cronExpression, diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index f67a7f92..04def5c0 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -25,6 +25,19 @@ spring: driver-class-name: org.h2.Driver username: sa password: + # Issue #50: explicit Hikari sizing. Default of 10 was being exhausted + # by O(100) cron jobs firing on minute boundaries (each run holds 3-4 + # connections sequentially) plus ChannelHealthMonitor's per-minute scan. + # 30 leaves headroom for HTTP / SSE / channel adapters even with the + # cron concurrency limiter (CronJobService.MAX_CONCURRENT_CRON_RUNS=8) + # at full saturation. + hikari: + maximum-pool-size: 30 + minimum-idle: 5 + connection-timeout: 30000 + idle-timeout: 600000 + max-lifetime: 1800000 + leak-detection-threshold: 60000 # SQL 初始化由 DatabaseBootstrapRunner 接管，关闭 Spring 自动执行 sql: diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V69__cron_job_dedup_unique.sql b/mateclaw-server/src/main/resources/db/migration/h2/V69__cron_job_dedup_unique.sql new file mode 100644 index 00000000..59f04e41 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V69__cron_job_dedup_unique.sql @@ -0,0 +1,20 @@ +-- Issue #50: deduplicate accumulated cron jobs and prevent future duplicates +-- at the DB level. +-- +-- Step 1 — purge duplicate active rows, keeping the earliest id (= earliest +-- creation, since IDs are snowflake-monotonic). Hard delete because this +-- entity has no @TableLogic; deleteById() already performs physical deletes. +DELETE FROM mate_cron_job +WHERE id NOT IN ( + SELECT keep_id FROM ( + SELECT MIN(id) AS keep_id + FROM mate_cron_job + GROUP BY workspace_id, agent_id, name + ) +); + +-- Step 2 — enforce uniqueness so concurrent LLM-driven creates can't +-- re-introduce duplicates. The service layer also dedups in-process; this +-- index is the racy-write safety net. +CREATE UNIQUE INDEX IF NOT EXISTS uk_cron_job_workspace_agent_name + ON mate_cron_job(workspace_id, agent_id, name); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V69__cron_job_dedup_unique.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V69__cron_job_dedup_unique.sql new file mode 100644 index 00000000..4d6ecfca --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V69__cron_job_dedup_unique.sql @@ -0,0 +1,25 @@ +-- Issue #50: deduplicate accumulated cron jobs and prevent future duplicates +-- at the DB level. MySQL doesn't allow DELETE with a subquery that scans the +-- same table directly, so we use the LEFT JOIN + IS NULL pattern. +-- +-- Step 1 — purge duplicate active rows, keeping the earliest id per +-- (workspace_id, agent_id, name). Hard delete because this entity has no +-- @TableLogic; deleteById() already performs physical deletes. +DELETE t FROM mate_cron_job t +LEFT JOIN ( + SELECT MIN(id) AS keep_id + FROM mate_cron_job + GROUP BY workspace_id, agent_id, name +) k ON t.id = k.keep_id +WHERE k.keep_id IS NULL; + +-- Step 2 — add the unique index, idempotent via INFORMATION_SCHEMA guard +-- (MySQL < 8.0.29 has no CREATE INDEX IF NOT EXISTS). +SET @idx_exists := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'mate_cron_job' + AND INDEX_NAME = 'uk_cron_job_workspace_agent_name'); +SET @stmt := IF(@idx_exists = 0, + 'CREATE UNIQUE INDEX uk_cron_job_workspace_agent_name ON mate_cron_job(workspace_id, agent_id, name)', + 'SELECT 1'); +PREPARE s FROM @stmt; EXECUTE s; DEALLOCATE PREPARE s;