From 94163f9afb82a26ffdf57af998e4d246f841b78a Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sat, 5 Sep 2026 22:31:12 +0800 Subject: [PATCH] =?UTF-8?q?fix=F0=9F=90=9B:=20stop=20the=20job=20scheduler?= =?UTF-8?q?=20on=20the=20way=20out,=20and=20start=20one=20per=20tenant?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-tenant setup ended with `defer crontab.Stop()` on the line above `select {}`. The select never returned, so the deferred call was unreachable for the life of the process: the scheduler had never once been stopped. And because setup never returned, the `for k, db := range dbs` loop in Setup never reached its second iteration - with several tenant databases configured, only whichever one came first out of the map ever got a scheduler at all. Both fall out of deleting the select, which was blocking for nothing: cron.Start is `go c.run()` and has never needed anything to hold the caller. The stop becomes a BeforeExit callback. cron.Stop returns a context that closes once the jobs already running have finished, so the shutdown budget has something real to bound - and giving up on that wait leaves those jobs running until the process exits, which is better than holding the whole shutdown open for one job that will not end. Startup moves from a bare goroutine in run() onto AfterListen. Two reasons: the phase runs behind core's panic guard, which does not reach across a goroutine boundary, so a panic while loading jobs used to take the process down; and the jobs it starts can call the API, which is only true once the socket is accepting. It can be synchronous now precisely because setup returns. Tested where it can be: startCrontab is split out so a scheduler can be started with no database in sight. The job runs every second; after RunShutdown, two and a half seconds of silence is the assertion. The counter-proof - registering no callback, which is what this commit replaces - compiles and reports "the job fired 2 more times after shutdown". There is one test, not several, because BeforeExit closes to further registration once it has run; a second RunShutdown in the same binary would find an empty registry and pass while proving nothing. **The multi-tenant half has no test.** setup needs a *gorm.DB per tenant before it reaches the line that was blocking, and this repository's CI has no database - `make build` is CGO_ENABLED=0 with no sqlite tag. It is the same defect though: the loop could not advance past a call that never returned. --- app/jobs/jobbase.go | 32 ++++++++++++++++++++++--- app/jobs/jobbase_test.go | 52 ++++++++++++++++++++++++++++++++++++++++ cmd/api/server.go | 25 ++++++++++++++----- 3 files changed, 100 insertions(+), 9 deletions(-) create mode 100644 app/jobs/jobbase_test.go 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("")