diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go
index d382172..88a6a1a 100644
--- a/internal/cmd/cmd.go
+++ b/internal/cmd/cmd.go
@@ -13,6 +13,7 @@ import (
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/gcmd"
+ "github.com/gogf/gf/v2/os/gctx"
"github.com/qianfree/team-api/internal/consts"
adminController "github.com/qianfree/team-api/internal/controller/admin"
@@ -335,9 +336,37 @@ func printBanner() {
fmt.Println()
}
+// channelAutoTestSchedule 按配置的探测间隔生成 cron 表达式。
+//
+// 用 @every 而非 */N：*/N 只在 N < 60 时有意义（*/90 在分钟位上等价于 */30，
+// 静默给出与配置不符的频率），而 @every 支持任意时长。副作用是触发时刻相对进程
+// 启动时间而非整点对齐——多实例部署时反而更好，不会所有实例卡在同一分钟同时探测。
+func channelAutoTestSchedule(ctx context.Context) string {
+ minutes := common.Config().GetInt(ctx, "channel_auto_test_interval_minutes")
+ // 配置缺失（GetInt 返回 0）或越界时回落默认值，避免非法表达式让任务注册失败
+ if minutes < 1 || minutes > 1440 {
+ minutes = 5
+ }
+ return fmt.Sprintf("@every %dm", minutes)
+}
+
+// watchChannelAutoTestInterval 探测间隔配置变更时重排 cron，无需重启。
+// 回调经 Redis pub/sub 触发，多实例会各自重排自己的调度器。
+func watchChannelAutoTestInterval(cs *common.CronScheduler) {
+ common.OnSettingsChanged(func(ctx context.Context, key string) {
+ if key != "channel_auto_test_interval_minutes" {
+ return
+ }
+ if err := cs.Reschedule(ctx, "channel_auto_test", channelAutoTestSchedule(ctx)); err != nil {
+ g.Log().Errorf(ctx, "渠道自动探测重排失败，沿用原间隔: %v", err)
+ }
+ })
+}
+
// registerCronJobs 集中注册所有定时任务，避免散落在主启动流程中。
// 新增 cron 任务时在此追加一项即可；任务名（用于分布式锁 key）需保持唯一且稳定。
func registerCronJobs(cs *common.CronScheduler) {
+ watchChannelAutoTestInterval(cs)
cs.Register("ops_system_collector", "系统指标采集", "* * * * *", func(ctx context.Context) error {
return monitor.CollectSystemMetrics(ctx)
})
@@ -353,7 +382,7 @@ func registerCronJobs(cs *common.CronScheduler) {
cs.Register("health_snapshot", "渠道健康快照", "*/5 * * * *", func(ctx context.Context) error {
return task.SnapshotHealthScores(ctx)
})
- cs.Register("channel_auto_test", "渠道自动测试", "*/5 * * * *", func(ctx context.Context) error {
+ cs.Register("channel_auto_test", "渠道自动测试", channelAutoTestSchedule(gctx.New()), func(ctx context.Context) error {
if common.Config().GetBool(ctx, "channel_auto_test_enabled") {
task.AutoTestChannels(ctx)
}
diff --git a/internal/logic/common/config.go b/internal/logic/common/config.go
index e648ed2..3c82cc7 100644
--- a/internal/logic/common/config.go
+++ b/internal/logic/common/config.go
@@ -387,6 +387,8 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) {
s.cache.Delete(ctx, "public_options")
s.cache.Delete(ctx, "category:"+getCategoryForKey(key))
g.Log().Debugf(ctx, "[PubSub:settings] 缓存已失效: %s", key)
+ // 缓存失效后再触发回调：回调内部大概率要读该配置的新值
+ fireSettingsChanged(ctx, key)
}
conn.Close(ctx)
@@ -394,6 +396,46 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) {
}()
}
+// ──────────────────────────────────────────
+// 配置变更回调
+// ──────────────────────────────────────────
+
+var (
+ settingsHookMu sync.RWMutex
+ settingsHooks []func(ctx context.Context, key string)
+)
+
+// OnSettingsChanged 注册配置变更回调，用于"改了配置要立刻做点什么"的场景
+// （如按新的探测间隔重排 cron），而不只是让缓存失效。
+//
+// 回调经 Redis pub/sub 触发，因此本实例自己的写入同样会走一遍（发布者也订阅了该频道），
+// 无需在写入路径上重复调用。Redis 不可用时回调不触发——与缓存失效同等降级。
+// 回调在订阅 goroutine 内串行执行，实现必须快速返回、自行兜底 panic。
+func OnSettingsChanged(fn func(ctx context.Context, key string)) {
+ settingsHookMu.Lock()
+ defer settingsHookMu.Unlock()
+ settingsHooks = append(settingsHooks, fn)
+}
+
+// fireSettingsChanged 依次触发已注册回调，单个回调 panic 不影响其余回调与订阅循环。
+func fireSettingsChanged(ctx context.Context, key string) {
+ settingsHookMu.RLock()
+ hooks := make([]func(ctx context.Context, key string), len(settingsHooks))
+ copy(hooks, settingsHooks)
+ settingsHookMu.RUnlock()
+
+ for _, fn := range hooks {
+ func() {
+ defer func() {
+ if r := recover(); r != nil {
+ g.Log().Errorf(ctx, "[PubSub:settings] 变更回调 panic (key=%s): %v", key, r)
+ }
+ }()
+ fn(ctx, key)
+ }()
+ }
+}
+
// ──────────────────────────────────────────
// Validation
// ──────────────────────────────────────────
diff --git a/internal/logic/common/cron.go b/internal/logic/common/cron.go
index acc0d2e..8ca7098 100644
--- a/internal/logic/common/cron.go
+++ b/internal/logic/common/cron.go
@@ -2,6 +2,7 @@ package common
import (
"context"
+ "runtime/debug"
"sync"
"time"
@@ -35,6 +36,10 @@ type CronScheduler struct {
running map[string]bool
runMu sync.Mutex
cron *cron.Cron
+ // entryIDs job 名 → robfig/cron 条目 ID，供 Reschedule 摘除旧条目。
+ // 仅在 StartBackground 之后有值。
+ entryIDs map[string]cron.EntryID
+ entryMu sync.Mutex
}
var (
@@ -60,8 +65,9 @@ func GetCronScheduler() *CronScheduler {
// NewCronScheduler creates a new CronScheduler.
func NewCronScheduler() *CronScheduler {
return &CronScheduler{
- running: make(map[string]bool),
- cron: cron.New(),
+ running: make(map[string]bool),
+ entryIDs: make(map[string]cron.EntryID),
+ cron: cron.New(),
}
}
@@ -180,7 +186,7 @@ func (cs *CronScheduler) runJobInternal(ctx context.Context, name, triggeredBy s
// Execute job
startTime := time.Now()
- handlerErr := job.Handler(ctx)
+ handlerErr := runHandlerSafely(ctx, job)
duration := time.Since(startTime)
status := "succeeded"
@@ -197,6 +203,23 @@ func (cs *CronScheduler) runJobInternal(ctx context.Context, name, triggeredBy s
return handlerErr
}
+// runHandlerSafely 执行 job handler 并兜底 panic。
+//
+// robfig/cron v3 的 cron.New() 默认不带 Recover 链，其 startJob 是裸的
+// `go func() { j.Run() }()`——任意一个 job panic 都会直接崩掉整个进程，而不只是
+// 让该任务失败。此处把 panic 转成普通 error：进程存活，且失败会经 recordExecution
+// 落进 sys_cron_jobs，管理后台的任务列表能看到，不会变成一次无声的重启。
+func runHandlerSafely(ctx context.Context, job *CronJob) (err error) {
+ defer func() {
+ if r := recover(); r != nil {
+ // 堆栈只进日志：last_error 会入库并在管理后台展示，塞完整堆栈会撑爆展示
+ g.Log().Errorf(ctx, "cron job panicked: %s: %v\n%s", job.Name, r, debug.Stack())
+ err = gerror.Newf("任务执行 panic: %v", r)
+ }
+ }()
+ return job.Handler(ctx)
+}
+
// recordExecution persists an execution record to the database.
// Always UPSERT sys_cron_jobs with latest status; no separate execution log table.
func (cs *CronScheduler) recordExecution(ctx context.Context, jobName, status string, startedAt time.Time, duration time.Duration, errMsg, triggeredBy string) {
@@ -268,7 +291,7 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
for _, job := range jobs {
j := job
- _, err := cs.cron.AddFunc(j.Schedule, func() {
+ entryID, err := cs.cron.AddFunc(j.Schedule, func() {
bgCtx := gctx.New()
if err := cs.RunJob(bgCtx, j.Name); err != nil {
g.Log().Errorf(bgCtx, "background cron error: %s: %v", j.Name, err)
@@ -276,7 +299,11 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
})
if err != nil {
g.Log().Errorf(ctx, "failed to register cron job %s with schedule %s: %v", j.Name, j.Schedule, err)
+ continue
}
+ cs.entryMu.Lock()
+ cs.entryIDs[j.Name] = entryID
+ cs.entryMu.Unlock()
}
cs.cron.Start()
@@ -291,6 +318,58 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
g.Log().Info(ctx, "cron scheduler started in background")
}
+// Reschedule 更新已注册 job 的调度表达式并即时生效（配置热更新用）。
+//
+// 先添加新条目、成功后再摘除旧条目：表达式非法时保持原调度不变，
+// 避免一次错误配置把任务从调度器里彻底摘掉。
+// StartBackground 之前调用无意义（尚无条目），此时只更新 job.Schedule 记录。
+func (cs *CronScheduler) Reschedule(ctx context.Context, name, schedule string) error {
+ cs.mu.RLock()
+ var job *CronJob
+ for _, j := range cs.jobs {
+ if j.Name == name {
+ job = j
+ break
+ }
+ }
+ cs.mu.RUnlock()
+ if job == nil {
+ return gerror.Newf("job %s not registered", name)
+ }
+
+ cs.entryMu.Lock()
+ defer cs.entryMu.Unlock()
+
+ oldID, started := cs.entryIDs[name]
+ if !started {
+ cs.mu.Lock()
+ job.Schedule = schedule
+ cs.mu.Unlock()
+ return nil
+ }
+
+ newID, err := cs.cron.AddFunc(schedule, func() {
+ bgCtx := gctx.New()
+ if err := cs.RunJob(bgCtx, name); err != nil {
+ g.Log().Errorf(bgCtx, "background cron error: %s: %v", name, err)
+ }
+ })
+ if err != nil {
+ return gerror.Wrapf(err, "invalid schedule %q for job %s", schedule, name)
+ }
+
+ cs.cron.Remove(oldID)
+ cs.entryIDs[name] = newID
+
+ cs.mu.Lock()
+ oldSchedule := job.Schedule
+ job.Schedule = schedule
+ cs.mu.Unlock()
+
+ g.Log().Infof(ctx, "cron job %s rescheduled: %s → %s", name, oldSchedule, schedule)
+ return nil
+}
+
// cronLockTTL 分布式锁过期时间：仅用于实例崩溃时的兜底自动释放（正常完成会主动释放）。
// 取值需大于绝大多数 job 的执行时长；过长则崩溃后会阻塞该 job 至多这么久。
const cronLockTTL = 10 * time.Minute
diff --git a/internal/logic/common/cron_test.go b/internal/logic/common/cron_test.go
new file mode 100644
index 0000000..ae5ff0e
--- /dev/null
+++ b/internal/logic/common/cron_test.go
@@ -0,0 +1,119 @@
+package common
+
+import (
+ "context"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// TestRunHandlerSafely_Panic转为错误 robfig/cron v3 的 cron.New() 不带 Recover 链，
+// startJob 是裸的 `go j.Run()`——任务 panic 会崩掉整个进程。此处必须兜住并转成
+// 普通 error，才能既保住进程、又让失败落进 sys_cron_jobs 供管理后台查看。
+func TestRunHandlerSafely_Panic转为错误(t *testing.T) {
+ job := &CronJob{
+ Name: "panicking_job",
+ Handler: func(_ context.Context) error {
+ panic("boom")
+ },
+ }
+
+ var err error
+ assert.NotPanics(t, func() {
+ err = runHandlerSafely(context.Background(), job)
+ }, "panic 必须被兜住，否则整个进程会崩")
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "boom", "错误信息应保留 panic 内容供排查")
+}
+
+func TestRunHandlerSafely_正常返回值透传(t *testing.T) {
+ ctx := context.Background()
+
+ okJob := &CronJob{Name: "ok", Handler: func(_ context.Context) error { return nil }}
+ assert.NoError(t, runHandlerSafely(ctx, okJob))
+
+ failJob := &CronJob{Name: "fail", Handler: func(_ context.Context) error {
+ return assert.AnError
+ }}
+ assert.ErrorIs(t, runHandlerSafely(ctx, failJob), assert.AnError, "普通错误必须原样返回")
+}
+
+// TestReschedule_未启动时只更新记录 StartBackground 之前没有 cron 条目，
+// 此时重排只改 job.Schedule，供随后的 StartBackground 按新表达式注册。
+func TestReschedule_未启动时只更新记录(t *testing.T) {
+ cs := NewCronScheduler()
+ cs.Register("job_a", "任务A", "*/5 * * * *", func(_ context.Context) error { return nil })
+
+ require.NoError(t, cs.Reschedule(context.Background(), "job_a", "@every 30m"))
+
+ jobs := cs.ListJobs()
+ require.Len(t, jobs, 1)
+ assert.Equal(t, "@every 30m", jobs[0].Schedule)
+}
+
+func TestReschedule_未注册任务报错(t *testing.T) {
+ cs := NewCronScheduler()
+ assert.Error(t, cs.Reschedule(context.Background(), "nonexistent", "@every 5m"))
+}
+
+// TestReschedule_非法表达式保持原调度 一次错误配置不能把任务从调度器里彻底摘掉，
+// 因此实现是「先加新条目、成功后再摘旧条目」。
+func TestReschedule_非法表达式保持原调度(t *testing.T) {
+ ctx := context.Background()
+ cs := NewCronScheduler()
+ cs.Register("job_b", "任务B", "@every 5m", func(_ context.Context) error { return nil })
+ cs.StartBackground(ctx)
+ defer cs.cron.Stop()
+
+ before := cs.ListJobs()[0].Schedule
+ require.Error(t, cs.Reschedule(ctx, "job_b", "不是合法表达式"))
+ assert.Equal(t, before, cs.ListJobs()[0].Schedule, "非法表达式不得改动原调度")
+
+ // 条目仍在调度器中，未被误摘
+ assert.Len(t, cs.cron.Entries(), 1)
+}
+
+// TestReschedule_有效表达式替换条目 同时验证 @every 语法确实被 cron.New() 的
+// 默认解析器接受——探测间隔配置正是靠它支持任意分钟数（*/N 在 N>=60 时会静默失真）。
+func TestReschedule_有效表达式替换条目(t *testing.T) {
+ ctx := context.Background()
+ cs := NewCronScheduler()
+ cs.Register("job_c", "任务C", "@every 5m", func(_ context.Context) error { return nil })
+ cs.StartBackground(ctx)
+ defer cs.cron.Stop()
+
+ require.Len(t, cs.cron.Entries(), 1)
+ oldID := cs.entryIDs["job_c"]
+
+ require.NoError(t, cs.Reschedule(ctx, "job_c", "@every 90m"))
+
+ assert.Equal(t, "@every 90m", cs.ListJobs()[0].Schedule)
+ assert.Len(t, cs.cron.Entries(), 1, "旧条目必须被摘除，否则会双倍触发")
+ assert.NotEqual(t, oldID, cs.entryIDs["job_c"], "条目 ID 应已更新")
+}
+
+// TestOnSettingsChanged_回调panic不影响其余回调 回调在订阅 goroutine 内串行执行，
+// 一个回调 panic 不能带崩订阅循环或阻断后续回调。
+func TestOnSettingsChanged_回调panic不影响其余回调(t *testing.T) {
+ settingsHookMu.Lock()
+ saved := settingsHooks
+ settingsHooks = nil
+ settingsHookMu.Unlock()
+ defer func() {
+ settingsHookMu.Lock()
+ settingsHooks = saved
+ settingsHookMu.Unlock()
+ }()
+
+ var firstCalled, thirdCalled bool
+ OnSettingsChanged(func(_ context.Context, _ string) { firstCalled = true })
+ OnSettingsChanged(func(_ context.Context, _ string) { panic("hook boom") })
+ OnSettingsChanged(func(_ context.Context, _ string) { thirdCalled = true })
+
+ assert.NotPanics(t, func() {
+ fireSettingsChanged(context.Background(), "some_key")
+ })
+ assert.True(t, firstCalled)
+ assert.True(t, thirdCalled, "前一个回调 panic 不得阻断后续回调")
+}
diff --git a/internal/logic/common/settings_registry.go b/internal/logic/common/settings_registry.go
index 67accf2..128f1bc 100644
--- a/internal/logic/common/settings_registry.go
+++ b/internal/logic/common/settings_registry.go
@@ -213,6 +213,9 @@ var Registry = []SettingDef{
// ── Channel ──
{Key: "channel_auto_test_enabled", Type: SettingTypeBool, Default: "true", Category: "channel",
Label: "渠道自动探测", Description: "定期向活跃渠道发送测试请求，检测连通性并更新健康度（会消耗少量 Token）。禁用渠道不自动探测，由管理员手动测试确认后再启用"},
+ {Key: "channel_auto_test_interval_minutes", Type: SettingTypeInt, Default: "5", Category: "channel",
+ Label: "自动探测间隔（分钟）", Validation: "min:1,max:1440",
+ Description: "两轮自动探测之间的间隔，保存后即时生效。探测是串行的（每渠道最多 30 秒），渠道多或上游超时多时单轮耗时会变长；上一轮未结束时下一轮自动跳过，因此间隔不宜小于「渠道数 × 30 秒」"},
{Key: "channel_auto_disable_enabled", Type: SettingTypeBool, Default: "false", Category: "channel",
Label: "渠道自动禁用", Description: "渠道熔断持续超过路由策略 breaker.autoDisableAfterSeconds（默认 10 分钟）未恢复时自动禁用"},
{Key: "channel_routing_policy", Type: SettingTypeJSON, Default: "", Category: "channel",
diff --git a/internal/logic/task/channel_auto_test_cron.go b/internal/logic/task/channel_auto_test_cron.go
index 8937f3c..6f5a29a 100644
--- a/internal/logic/task/channel_auto_test_cron.go
+++ b/internal/logic/task/channel_auto_test_cron.go
@@ -43,18 +43,7 @@ func AutoTestChannels(ctx context.Context) {
failCount := 0
for _, ch := range channels {
- testCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
- result, err := admin.New().TestChannel(testCtx, &v1.ChannelTestReq{ID: ch.ID})
- cancel()
-
- if err != nil {
- failCount++
- // 探测请求尚未发出（无可用 Key/渠道不存在等），TestChannel 内部无结果日志，此处记录
- g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测执行出错: %v", ch.Name, ch.ID, err)
- continue
- }
-
- if result.Success {
+ if probeOneChannel(ctx, ch.ID, ch.Name) {
successCount++
} else {
failCount++
@@ -65,3 +54,27 @@ func AutoTestChannels(ctx context.Context) {
g.Log().Infof(ctx, "[ChannelProbe] 自动探测完成: 共 %d 个渠道 | 成功 %d | 失败 %d",
len(channels), successCount, failCount)
}
+
+// probeOneChannel 探测单个渠道，返回是否成功。
+// 独立超时 + panic 兜底：单个渠道无论超时、报错还是 panic，都不影响本轮后续渠道
+// （panic 若逃逸到 cron 层虽有 runHandlerSafely 兜底进程，但本轮剩余渠道会被整体跳过）。
+func probeOneChannel(ctx context.Context, channelID int64, channelName string) (ok bool) {
+ defer func() {
+ if r := recover(); r != nil {
+ ok = false
+ g.Log().Errorf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测 panic，跳过该渠道继续下一个: %v",
+ channelName, channelID, r)
+ }
+ }()
+
+ testCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+ defer cancel()
+
+ result, err := admin.New().TestChannel(testCtx, &v1.ChannelTestReq{ID: channelID})
+ if err != nil {
+ // 探测请求尚未发出（无可用 Key/渠道不存在等），TestChannel 内部无结果日志，此处记录
+ g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测执行出错: %v", channelName, channelID, err)
+ return false
+ }
+ return result.Success
+}
diff --git a/web/admin/src/views/system/settings/ChannelTab.vue b/web/admin/src/views/system/settings/ChannelTab.vue
index 614f726..40e1628 100644
--- a/web/admin/src/views/system/settings/ChannelTab.vue
+++ b/web/admin/src/views/system/settings/ChannelTab.vue
@@ -16,6 +16,19 @@ const values = useFormValues()
@change="(v: string | number | boolean) => values['channel_auto_test_enabled'] = v"
/>
+
+
+ 保存后即时生效，无需重启。探测是串行的（每渠道最多 30 秒），渠道多或上游超时多时单轮耗时会变长；
+ 上一轮未结束时下一轮自动跳过，因此间隔不宜小于「渠道数 × 30 秒」。
+