diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java index 5ac67cd6..e5ab9bf9 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerEventIngestService.java @@ -2,8 +2,11 @@ package vip.mate.trigger.ingest; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.dao.DuplicateKeyException; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import vip.mate.trigger.dispatch.DispatchResult; import vip.mate.trigger.dispatch.TriggerDispatcher; @@ -59,6 +62,26 @@ public class TriggerEventIngestService { private final TriggerPatternMatcher patternMatcher; private final TriggerRateLimiter rateLimiter = new TriggerRateLimiter(); + /** When true (production default), {@code dispatcher.dispatch} runs on + * a worker thread so the caller (webhook / scheduler / runner) returns + * quickly. When false, ingest runs the workflow inline on the caller + * thread; tests pin to false so they can assert against downstream + * workflow state immediately after {@code ingest()} returns. */ + @Value("${mateclaw.workflow.trigger.async-dispatch:true}") + private boolean asyncDispatch; + + @Value("${mateclaw.workflow.trigger.dispatch-pool-size:8}") + private int dispatchPoolSize; + + @Value("${mateclaw.workflow.trigger.dispatch-queue-capacity:256}") + private int dispatchQueueCapacity; + + /** Lazy-built bounded thread pool used when {@link #asyncDispatch} is + * true. CallerRunsPolicy is the back-pressure: when the queue is full + * the calling thread runs the dispatch itself, which guarantees no + * silent drop while still capping in-flight work. */ + private volatile java.util.concurrent.ThreadPoolExecutor dispatchExecutor; + public TriggerEventIngestService(TriggerMapper triggerMapper, TriggerEventMapper eventMapper, TriggerDispatcher dispatcher, @@ -73,6 +96,44 @@ public class TriggerEventIngestService { this.patternMatcher = patternMatcher; } + private java.util.concurrent.ThreadPoolExecutor dispatchExecutor() { + java.util.concurrent.ThreadPoolExecutor local = dispatchExecutor; + if (local != null) return local; + synchronized (this) { + if (dispatchExecutor == null) { + int size = Math.max(1, dispatchPoolSize); + int cap = Math.max(1, dispatchQueueCapacity); + dispatchExecutor = new java.util.concurrent.ThreadPoolExecutor( + size, size, + 60L, java.util.concurrent.TimeUnit.SECONDS, + new java.util.concurrent.LinkedBlockingQueue<>(cap), + r -> { + Thread t = new Thread(r, "trigger-dispatch-" + System.currentTimeMillis()); + t.setDaemon(true); + return t; + }, + new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy()); + } + return dispatchExecutor; + } + } + + @PreDestroy + void shutdownDispatchExecutor() { + java.util.concurrent.ThreadPoolExecutor local = dispatchExecutor; + if (local != null) { + local.shutdown(); + try { + if (!local.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) { + local.shutdownNow(); + } + } catch (InterruptedException e) { + local.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + } + /** * Process one envelope through the pipeline. Returns a result per * candidate trigger so callers can surface a partial-accept summary. @@ -119,19 +180,29 @@ public class TriggerEventIngestService { if (!rateLimiter.tryAcquire(trigger.getId(), limit, Instant.now())) { return IngestResult.dropped(trigger.getId(), Reason.RATE_LIMITED); } - DispatchResult outcome; - try { - outcome = dispatcher.dispatch(trigger, envelope.data()); - } catch (Exception e) { - // Belt-and-suspenders — the dispatcher already wraps its own - // exceptions, but if anything escapes we mark it as DISPATCH_ERROR - // and persist last_error so the UI surfaces *why*. - log.error("Trigger {} dispatch threw on event ingest: {}", - trigger.getId(), e.getMessage(), e); - persistDispatchOutcome(trigger, DispatchResult.failed("dispatch threw: " + e.getMessage())); - return IngestResult.dropped(trigger.getId(), Reason.DISPATCH_ERROR); + if (asyncDispatch) { + // Async path — submit dispatch to the bounded pool so the + // caller (webhook / scheduler / runner thread) returns + // quickly. Bookkeeping happens inside the worker, so + // last_error / fireCount stay accurate. The IngestResult + // signals "accepted, fanning out" rather than "ran to + // completion"; that's the honest contract for an async + // pipeline. CallerRunsPolicy on the executor means we + // self-throttle instead of dropping under back-pressure. + try { + dispatchExecutor().execute(() -> runDispatchAndPersist(trigger, envelope)); + } catch (Exception e) { + log.error("Trigger {} dispatch submit failed: {}", + trigger.getId(), e.getMessage(), e); + persistDispatchOutcome(trigger, + DispatchResult.failed("dispatch submit failed: " + e.getMessage())); + return IngestResult.dropped(trigger.getId(), Reason.DISPATCH_ERROR); + } + return IngestResult.fired(trigger.getId()); } - persistDispatchOutcome(trigger, outcome); + // Synchronous path — used by tests and any deployment that + // explicitly opts out via mateclaw.workflow.trigger.async-dispatch=false. + DispatchResult outcome = runDispatchAndPersist(trigger, envelope); return switch (outcome.kind()) { case FIRED -> IngestResult.fired(trigger.getId()); case SKIPPED -> IngestResult.dropped(trigger.getId(), Reason.DISPATCH_SKIPPED); @@ -139,6 +210,22 @@ public class TriggerEventIngestService { }; } + /** Runs dispatch + bookkeeping on whatever thread invokes it (the + * caller in sync mode, a worker in async mode). Returns the + * outcome so sync callers can map it back to an IngestResult. */ + private DispatchResult runDispatchAndPersist(TriggerEntity trigger, TriggerEventEnvelope envelope) { + DispatchResult outcome; + try { + outcome = dispatcher.dispatch(trigger, envelope.data()); + } catch (Exception e) { + log.error("Trigger {} dispatch threw on event ingest: {}", + trigger.getId(), e.getMessage(), e); + outcome = DispatchResult.failed("dispatch threw: " + e.getMessage()); + } + persistDispatchOutcome(trigger, outcome); + return outcome; + } + /** * Update the trigger row's bookkeeping based on the dispatch outcome. * Only FIRED bumps {@code fireCount} and {@code lastFiredAt} — SKIPPED @@ -210,6 +297,28 @@ public class TriggerEventIngestService { LocalDateTime.ofInstant(Instant.now(), ZoneOffset.systemDefault()))); } + /** + * Periodic sweep of expired {@code mate_trigger_event} dedup rows. + * Default cadence is every 5 minutes, tunable via + * {@code mateclaw.workflow.trigger.dedup-sweep-interval-ms}. The + * initial delay matches the cadence so a JVM that just started doesn't + * race {@code recordDedupRow} for the same window. + */ + @Scheduled( + fixedDelayString = "${mateclaw.workflow.trigger.dedup-sweep-interval-ms:300000}", + initialDelayString = "${mateclaw.workflow.trigger.dedup-sweep-initial-delay-ms:300000}") + public void scheduledSweepExpired() { + try { + int dropped = sweepExpired(); + if (dropped > 0) { + log.info("[TriggerIngest] swept {} expired dedup rows", dropped); + } + } catch (Exception e) { + // Best-effort — never let the sweep crash the scheduler thread. + log.warn("[TriggerIngest] dedup sweep failed: {}", e.getMessage()); + } + } + public enum Reason { PATTERN_MISMATCH, BOT_SELF, DUPLICATE, RATE_LIMITED, EXHAUSTED, /** Dispatcher returned SKIPPED — pre-flight rejected (no published revision, etc.). */ diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java index 31f6ed6e..e657b543 100644 --- a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java @@ -2,7 +2,9 @@ package vip.mate.workflow.runtime; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import vip.mate.workflow.model.WorkflowPayloadEntity; import vip.mate.workflow.repository.WorkflowPayloadMapper; @@ -15,6 +17,7 @@ import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.time.LocalDateTime; import java.util.HexFormat; +import java.util.List; import java.util.Objects; import java.util.UUID; @@ -38,6 +41,7 @@ import java.util.UUID; * v1. The fs tier is what unblocks local dev / docker / private deploys * that don't have an object store configured. */ +@Slf4j @Service public class PayloadStore { @@ -50,17 +54,20 @@ public class PayloadStore { private final long inlineMaxBytes; private final long hardCapBytes; private final Path fsRoot; + private final long retentionDays; public PayloadStore(WorkflowPayloadMapper payloadMapper, ObjectMapper objectMapper, @Value("${mateclaw.workflow.payload.inline-max-bytes:262144}") long inlineMaxBytes, @Value("${mateclaw.workflow.payload.hard-cap-bytes:52428800}") long hardCapBytes, - @Value("${mateclaw.workflow.payload.fs.root:./data/workflow-payload}") String fsRoot) { + @Value("${mateclaw.workflow.payload.fs.root:./data/workflow-payload}") String fsRoot, + @Value("${mateclaw.workflow.payload.retention-days:30}") long retentionDays) { this.payloadMapper = payloadMapper; this.objectMapper = objectMapper; this.inlineMaxBytes = inlineMaxBytes; this.hardCapBytes = hardCapBytes; this.fsRoot = Path.of(fsRoot).toAbsolutePath(); + this.retentionDays = retentionDays; } /** Store a UTF-8 string payload and return its stable URI. */ @@ -172,6 +179,71 @@ public class PayloadStore { } } + /** + * Drop payload rows older than {@code retention-days}. Tombstones the + * filesystem files for fs-tier payloads in the same pass so the disk + * doesn't keep growing once the DB row is gone. Returns the number of + * rows actually deleted; primarily for tests + log lines. + * + *

v0 deletes by absolute age rather than walking the + * {@code mate_workflow_run} graph — runs that finish stay queryable + * for {@code retention-days} from the payload-write timestamp, which + * is "good enough" for an alpha. v1 can switch to run-state-driven + * GC ({@code state IN ('succeeded','failed') AND completed_at < + * threshold}) once the operator UI exposes a "preserve forever" flag + * for runs the customer wants kept. + */ + public int sweepExpired() { + if (retentionDays <= 0) return 0; + LocalDateTime cutoff = LocalDateTime.now().minusDays(retentionDays); + List stale = payloadMapper.selectList( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .lt(WorkflowPayloadEntity::getCreatedAt, cutoff)); + if (stale.isEmpty()) return 0; + int deleted = 0; + for (WorkflowPayloadEntity row : stale) { + // Best-effort fs cleanup before the row goes — the row IS the + // foreign key the file is reachable through; if the row goes + // first the file becomes orphaned. + if (STORAGE_KIND_FS.equals(row.getStorageKind()) && row.getStorageRef() != null) { + try { + Files.deleteIfExists(fsRoot.resolve(row.getStorageRef())); + } catch (IOException e) { + log.warn("[PayloadStore] fs delete failed for {}: {}", + row.getStorageRef(), e.getMessage()); + } + } + try { + payloadMapper.deleteById(row.getId()); + deleted++; + } catch (Exception e) { + log.warn("[PayloadStore] db delete failed for payload {}: {}", + row.getPayloadUri(), e.getMessage()); + } + } + return deleted; + } + + /** + * Periodic sweep — runs once an hour by default. Tunable via + * {@code mateclaw.workflow.payload.sweep-interval-ms}. Skips a tick + * silently when retentionDays = 0 (operator opted out of GC). + */ + @Scheduled( + fixedDelayString = "${mateclaw.workflow.payload.sweep-interval-ms:3600000}", + initialDelayString = "${mateclaw.workflow.payload.sweep-initial-delay-ms:600000}") + public void scheduledSweepExpired() { + try { + int dropped = sweepExpired(); + if (dropped > 0) { + log.info("[PayloadStore] swept {} expired payload rows (retention={} days)", + dropped, retentionDays); + } + } catch (Exception e) { + log.warn("[PayloadStore] periodic sweep failed: {}", e.getMessage()); + } + } + /** Wrapper exception for payload-store failures. */ public static class PayloadStoreException extends RuntimeException { public PayloadStoreException(String message) { super(message); } diff --git a/mateclaw-ui/src/api/index.ts b/mateclaw-ui/src/api/index.ts index e4db5ef2..0229fc38 100644 --- a/mateclaw-ui/src/api/index.ts +++ b/mateclaw-ui/src/api/index.ts @@ -911,6 +911,13 @@ export interface TriggerSummary { fireCount: number maxFires: number lastFiredAt?: string + /** Stamp of the last dispatch attempt regardless of outcome (FIRED / + * SKIPPED / FAILED). Distinguishes "never attempted" from + * "attempted but the pre-flight skipped". */ + lastDispatchedAt?: string + /** Most recent dispatch outcome message; null when the last attempt + * fired cleanly. */ + lastError?: string patternVersion: number createTime: string updateTime: string diff --git a/mateclaw-ui/src/i18n/locales/en-US.ts b/mateclaw-ui/src/i18n/locales/en-US.ts index ed6e3fae..ad641899 100644 --- a/mateclaw-ui/src/i18n/locales/en-US.ts +++ b/mateclaw-ui/src/i18n/locales/en-US.ts @@ -2077,6 +2077,15 @@ export default { }, targetWorkflowSelect: 'Pick a published workflow', targetWorkflowEmpty: 'No published workflows yet.', + lastDispatchedAt: 'Last attempt', + patternTypeLabels: { + cron: 'Cron schedule', + channel_message: 'Channel message', + content_match: 'Content match', + agent_lifecycle: 'Agent lifecycle', + workflow_completion: 'Workflow completion', + webhook: 'Webhook', + }, patternHints: { cron: 'Example: {"cron":"0 0 * * * *","timezone":"UTC"} — every cron change bumps pattern_version.', channel_message: 'Optional channelType (e.g. feishu) and senderEquals narrow the match.', diff --git a/mateclaw-ui/src/i18n/locales/zh-CN.ts b/mateclaw-ui/src/i18n/locales/zh-CN.ts index efa8f08e..e1ca59b4 100644 --- a/mateclaw-ui/src/i18n/locales/zh-CN.ts +++ b/mateclaw-ui/src/i18n/locales/zh-CN.ts @@ -2089,6 +2089,15 @@ export default { }, targetWorkflowSelect: '选择已发布的工作流', targetWorkflowEmpty: '尚无已发布的工作流。', + lastDispatchedAt: '最近一次尝试', + patternTypeLabels: { + cron: '定时任务', + channel_message: '渠道消息', + content_match: '内容匹配', + agent_lifecycle: '智能体生命周期', + workflow_completion: '工作流完成', + webhook: 'Webhook', + }, patternHints: { cron: '示例：{"cron":"0 0 * * * *","timezone":"Asia/Shanghai"} —— 每次修改 cron 都会让 pattern_version 自增。', channel_message: '可填 channelType（如 feishu）和 senderEquals 进一步过滤。', diff --git a/mateclaw-ui/src/views/Triggers.vue b/mateclaw-ui/src/views/Triggers.vue index bdeadf9d..f99e1542 100644 --- a/mateclaw-ui/src/views/Triggers.vue +++ b/mateclaw-ui/src/views/Triggers.vue @@ -27,16 +27,28 @@ - {{ row.name || t('triggers.unnamed') }} - {{ row.patternType }} -

{{ row.patternJson }}
+
{{ row.name || t('triggers.unnamed') }}
+
+ ⚠ {{ truncateError(row.lastError) }} +
+ + + {{ patternTypeLabel(row.patternType) }} +
{{ patternSummary(row) }}
{{ formatTarget(row) }} {{ t('triggers.rateUnit', { count: row.rateLimitPerMin }) }} {{ row.fireCount }} / {{ row.maxFires }} {{ row.patternVersion }} - {{ formatTime(row.lastFiredAt) }} + +
{{ formatTime(row.lastFiredAt) }}
+
+ {{ formatTime(row.lastDispatchedAt) }} +
+