diff --git a/app/jobs/jobbase.go b/app/jobs/jobbase.go index d1f5aba9..19fab0f3 100644 --- a/app/jobs/jobbase.go +++ b/app/jobs/jobbase.go @@ -1,6 +1,7 @@ package jobs import ( + "context" "fmt" log "github.com/go-admin-team/go-admin-core/v2/logger" "github.com/go-admin-team/go-admin-core/v2/sdk" @@ -145,11 +146,36 @@ func setup(key string, db *gorm.DB) { } // 其中任务 - crontab.Start() + startCrontab(crontab) +} + +// startCrontab starts c and arranges for it to be stopped on the way out. +// +// The stop used to be `defer crontab.Stop()` followed by `select {}`. The +// select never returned, so the defer never ran and the scheduler was never +// stopped; and because setup never returned, the loop in Setup never reached +// the second tenant - only whichever database came first out of the map ever +// got a scheduler at all. cron.Start is itself `go c.run()`, so the select was +// blocking for nothing. +// +// cron.Stop returns a context that closes once the jobs already running have +// finished. That is the wait the shutdown budget exists to bound: giving up on +// it leaves those jobs running until the process exits, which is better than +// holding the whole shutdown open for one job that will not end. +func startCrontab(c *cron.Cron) { + c.Start() fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore start success.") + // 关闭任务 - defer crontab.Stop() - select {} + sdk.Runtime.SetShutdown(func(ctx context.Context) { + stopped := c.Stop() + select { + case <-stopped.Done(): + fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore stopped.") + case <-ctx.Done(): + fmt.Println(time.Now().Format(timeFormat), " [WARN] JobCore stop gave up waiting for running jobs") + } + }) } // AddJob 添加任务 AddJob(invokeTarget string, jobId int, jobName string, cronExpression string) diff --git a/app/jobs/jobbase_test.go b/app/jobs/jobbase_test.go new file mode 100644 index 00000000..0d7c0028 --- /dev/null +++ b/app/jobs/jobbase_test.go @@ -0,0 +1,52 @@ +package jobs + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/pkg/cronjob" +) + +// The scheduler had never been stopped. `defer crontab.Stop()` sat directly +// above a `select {}` that never returned, so the deferred call was +// unreachable for the life of the process. +// +// There is one test rather than several because BeforeExit closes to further +// registration once it has run: a second RunShutdown in this binary would find +// an empty registry and pass while proving nothing. +func TestTheSchedulerIsStoppedOnTheWayOut(t *testing.T) { + var ticks atomic.Int64 + + c := cronjob.NewWithSeconds() + if _, err := c.AddFunc("* * * * * *", func() { ticks.Add(1) }); err != nil { + t.Fatalf("AddFunc: %v", err) + } + + startCrontab(c) + + // It has to be running before stopping it can mean anything. + deadline := time.Now().Add(5 * time.Second) + for ticks.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if ticks.Load() == 0 { + t.Fatal("the scheduler never ran the job, so this test cannot show it was stopped") + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := sdk.Runtime.RunShutdown(ctx); err != nil { + t.Fatalf("RunShutdown: %v", err) + } + + // Two and a half seconds is two more firings of a job that runs every + // second, so silence here is the assertion. + at := ticks.Load() + time.Sleep(2500 * time.Millisecond) + if n := ticks.Load() - at; n > 0 { + t.Errorf("the job fired %d more times after shutdown: the scheduler is still running", n) + } +} diff --git a/cmd/api/server.go b/cmd/api/server.go index ca510096..6829ca98 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -72,6 +72,13 @@ func setup() { // queue would have no consumers until somebody edited the config file. sdk.Runtime.SetPhase(runtime.AfterResource, attachQueueConsumers) + // On AfterListen rather than on a bare goroutine from run(). Two reasons: + // the phase runs behind core's panic guard, which does not reach across a + // goroutine boundary - a panic while loading jobs used to take the whole + // process down with a stack that named this file - and the jobs it starts + // can call the API, which is only true once the socket is accepting. + sdk.Runtime.SetPhase(runtime.AfterListen, startCronJobs) + //1. 读取配置 bootstrap.SetupConfig( file.NewSource(file.WithPath(configYml)), @@ -83,6 +90,18 @@ func setup() { log.Info(usageStr) } +// startCronJobs registers the job implementations and starts a scheduler for +// every tenant database. +// +// It is synchronous, like the phase that runs it. jobs.Setup returns now that +// the `select {}` at the end of its per-tenant setup is gone, which is what +// makes that possible; while it was there this could only be a goroutine, and +// a goroutine is outside the panic guard. +func startCronJobs() { + jobs.InitJob() + jobs.Setup(sdk.Runtime.GetAllDb()) +} + // attachedQueue is the queue generation the consumers are attached to, plus // one, so that the zero value means "attached to nothing yet". Written from // the goroutine running the phase, read from the next one - rounds never @@ -157,12 +176,6 @@ func run() error { WriteTimeout: time.Duration(config.ApplicationConfig.WriterTimeout) * time.Second, } - go func() { - jobs.InitJob() - jobs.Setup(sdk.Runtime.GetAllDb()) - - }() - if apiCheck { var routers = sdk.Runtime.GetRouter() q := sdk.Runtime.GetQueuePrefix("")