fix(cron): dedup scheduled jobs + connection pool guard (issue #50)

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<CronJobEntity>()
+                        .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(() -> {
+        cronExecutor.submit(() -> runWithBackpressure(entity, "manual"));
-            try {
-                cronJobRunner.executeJob(entity, "manual");
-            } finally {
-                updateRunTimes(entity.getId(), entity.getCronExpression(), entity.getTimezone());
-            }
-        });
     }
 
     // ==================== 调度器管理 ====================
@@ -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(() -> {
+                    cronExecutor.submit(() -> runWithBackpressure(job, "scheduled")), trigger);
-                        try {
-                            cronJobRunner.executeJob(job, "scheduled");
-                        } finally {
-                            updateRunTimes(job.getId(), job.getCronExpression(), job.getTimezone());
-                        }
-                    }), 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.
+     *
+     * <p>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
      */

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,

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:

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

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;