diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index bf344f53..5fe7b7a7 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -68,6 +68,28 @@ jobs: --health-retries 20 --health-start-period 20s + # The dialect most installations actually run, and until the + # scheduler lease (#915) the only one with no service here. The lease + # reads the database's clock, and the first implementation read it as + # a timestamp: over go-admin's own `parseTime=True&loc=Local` DSN, + # MySQL's UTC_TIMESTAMP comes back relabelled as local time, so on any + # host that is not UTC every lease was one zone offset out - and every + # assertion that compared the lease only against itself still passed. + # The three dialects that were here could not see it. + mysql: + image: mysql:8 + env: + MYSQL_ROOT_PASSWORD: GoAdmin_Test1 + MYSQL_DATABASE: goadmin_test + ports: + - 3306:3306 + options: >- + --health-cmd "mysqladmin ping -h 127.0.0.1 -uroot -pGoAdmin_Test1" + --health-interval 10s + --health-timeout 5s + --health-retries 20 + --health-start-period 20s + env: GO_ADMIN_TEST_REDIS_ADDR: 127.0.0.1:6379 # The soft-delete conversion drops an index, and gorm's PostgreSQL driver @@ -77,6 +99,11 @@ jobs: # See go-admin#919. GO_ADMIN_TEST_POSTGRES_DSN: "host=127.0.0.1 port=5432 user=postgres password=postgres dbname=goadmin_test sslmode=disable" GO_ADMIN_TEST_SQLSERVER_DSN: "sqlserver://sa:GoAdmin_Test1@127.0.0.1:1433?database=goadmin_test" + # loc=Local on purpose: it is what config/settings.yml ships and what + # made the timezone defect above reachable. A DSN here that quietly + # differed from the one installations use would test a configuration + # nobody runs. + GO_ADMIN_TEST_MYSQL_DSN: "root:GoAdmin_Test1@tcp(127.0.0.1:3306)/goadmin_test?charset=utf8mb4&parseTime=True&loc=Local" steps: diff --git a/app/jobs/jobbase.go b/app/jobs/jobbase.go index 49561ebd..5cbb8549 100644 --- a/app/jobs/jobbase.go +++ b/app/jobs/jobbase.go @@ -97,13 +97,20 @@ LOOP: } // Setup 初始化 +// Setup gives every tenant a scheduler and a supervisor to decide whether +// this instance is the one that fills it. +// +// One owner id for the whole process, not one per tenant: the thing holding +// the leases is this process, and a log line naming it should name the same +// thing in every database it appears in. func Setup(dbs map[string]*gorm.DB) { fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore Starting...") + owner := newOwnerID() for k, db := range dbs { sdk.Runtime.SetCrontabByTenant(k, cronjob.NewWithSeconds()) - setup(k, db) + newSupervisor(k, db, owner).start() } } @@ -149,7 +156,7 @@ func setup(key string, db *gorm.DB) { startCrontab(crontab) } -// startCrontab starts c and arranges for it to be stopped on the way out. +// startCrontab starts c. // // 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 @@ -158,24 +165,36 @@ func setup(key string, db *gorm.DB) { // 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. +// Stopping is no longer arranged here. A scheduler now stops for two +// different reasons - the process is going down, or this instance lost the +// lease (#915) - and only the supervisor knows which. Registering a shutdown +// callback per start, when a start happens every time the lease is taken, +// would also add one callback per leadership change for the life of the +// process: SetShutdown appends. func startCrontab(c *cron.Cron) { c.Start() fmt.Println(time.Now().Format(timeFormat), " [INFO] JobCore start success.") +} - // 关闭任务 - 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") - } - }) +// stopCrontab stops one tenant's scheduler and waits for the jobs already +// running to finish, bounded by ctx. +// +// cron.Stop returns a context that closes once those jobs have finished. +// That is the wait the shutdown budget exists to bound: giving up on it +// leaves them running until the process exits, which is better than holding +// the whole shutdown open for one job that will not end. +func stopCrontab(ctx context.Context, key string) { + c := sdk.Runtime.GetCrontabByTenant(key) + if c == nil { + return + } + 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 index 0d7c0028..175f6123 100644 --- a/app/jobs/jobbase_test.go +++ b/app/jobs/jobbase_test.go @@ -6,26 +6,45 @@ import ( "testing" "time" + glebarez "github.com/glebarez/sqlite" "github.com/go-admin-team/go-admin-core/v2/sdk" "github.com/go-admin-team/go-admin-core/v2/sdk/pkg/cronjob" + "gorm.io/gorm" + + models2 "go-admin/app/jobs/models" ) // 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. // +// It now goes through the supervisor, which is what production does and what +// owns the shutdown callback since the lease landed (#915): a scheduler stops +// either because the process is going down or because this instance lost the +// lease, and only the supervisor can tell those apart. +// // 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) { +// registration once it has run: a second RunShutdown in this binary would +// find an empty registry and pass while proving nothing. The lease-release +// assertion is folded in here for the same reason. +func TestTheSchedulerIsStoppedAndTheLeaseHandedBackOnTheWayOut(t *testing.T) { + const tenant = "*" + + db := leaseDB(t) var ticks atomic.Int64 c := cronjob.NewWithSeconds() if _, err := c.AddFunc("* * * * * *", func() { ticks.Add(1) }); err != nil { t.Fatalf("AddFunc: %v", err) } + sdk.Runtime.SetCrontabByTenant(tenant, c) - startCrontab(c) + s := newSupervisor(tenant, db, "instance-under-test") + s.start() + + if !s.holdsLease() { + t.Fatal("the supervisor did not take a free lease, so this test would prove nothing about giving it back") + } // It has to be running before stopping it can mean anything. deadline := time.Now().Add(5 * time.Second) @@ -49,4 +68,33 @@ func TestTheSchedulerIsStoppedOnTheWayOut(t *testing.T) { if n := ticks.Load() - at; n > 0 { t.Errorf("the job fired %d more times after shutdown: the scheduler is still running", n) } + + // And the lease is free, so a successor takes it immediately instead of + // waiting out a TTL held by a process that has exited. + var row models2.SysJobLease + if err := db.Where("name = ?", models2.SchedulerLeaseName).First(&row).Error; err != nil { + t.Fatalf("reading the lease row: %v", err) + } + if row.Owner != "" { + t.Errorf("the lease is still owned by %q after shutdown; a successor would wait out the TTL", row.Owner) + } +} + +// leaseDB is a database with the two tables jobs.setup touches and one free +// lease row, which is the shape migration 1786700009000 leaves behind. +func leaseDB(t *testing.T) *gorm.DB { + t.Helper() + + db, err := gorm.Open(glebarez.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatalf("opening sqlite: %v", err) + } + if err := db.AutoMigrate(&models2.SysJob{}, &models2.SysJobLease{}); err != nil { + t.Fatalf("migrating: %v", err) + } + row := models2.SysJobLease{Name: models2.SchedulerLeaseName} + if err := db.Create(&row).Error; err != nil { + t.Fatalf("seeding the lease row: %v", err) + } + return db } diff --git a/app/jobs/lease.go b/app/jobs/lease.go new file mode 100644 index 00000000..3f896914 --- /dev/null +++ b/app/jobs/lease.go @@ -0,0 +1,177 @@ +package jobs + +import ( + "fmt" + "os" + "time" + + "github.com/google/uuid" + "gorm.io/gorm" + + models2 "go-admin/app/jobs/models" +) + +// nowExprMs is the dialect's expression for the current time as +// milliseconds since the Unix epoch. +// +// The lease compares one instance's idea of "expired" against another +// instance's idea of "still mine", so both have to come from the same clock. +// Two processes whose wall clocks differ by more than the lease TTL would +// otherwise both hold it and both schedule - the exact situation the lease +// exists to prevent, and it would look like it was working, because each +// instance's own arithmetic is self-consistent. +// +// Milliseconds rather than a timestamp, because a timestamp does not survive +// the trip through a driver unchanged. MySQL's UTC_TIMESTAMP read over +// go-admin's own `parseTime=True&loc=Local` DSN arrives labelled as local +// time: on a UTC+8 host every lease is eight hours out, and a test that only +// checked the lease logic against itself passes anyway. An epoch integer has +// no timezone for a driver to apply. +func nowExprMs(dialect string) (string, error) { + switch dialect { + case "mysql": + // UNIX_TIMESTAMP reads its argument in the session timezone and + // NOW(3) is in the session timezone, so the two cancel and the + // result is the absolute epoch regardless of what that zone is. + return "CAST(ROUND(UNIX_TIMESTAMP(NOW(3)) * 1000) AS SIGNED)", nil + case "postgres": + return "CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)", nil + case "sqlite": + // julianday is the portable millisecond clock here: strftime('%s') + // truncates to the second, and unixepoch('now','subsec') needs + // SQLite 3.42. + return "CAST((julianday('now') - 2440587.5) * 86400000.0 AS INTEGER)", nil + case "sqlserver": + return "DATEDIFF_BIG(millisecond, '1970-01-01T00:00:00', SYSUTCDATETIME())", nil + } + return "", fmt.Errorf("no epoch-milliseconds expression for dialect %q", dialect) +} + +// dbNowMs reads the clock from the database rather than from this process. +// +// The read and the UPDATE that uses it are two statements, so the value is +// already slightly stale by the time it is compared - and that is the safe +// direction in both places it is used: +// +// - as the expiry cutoff, a stale-old now makes this instance *less* +// likely to decide another instance's lease has expired; +// - as the basis for a new expiry, it makes this instance's own lease +// expire sooner, so it renews sooner. +// +// Neither error makes two instances hold the lease at once. +// +// Zero is rejected rather than returned. It is what a failed conversion +// looks like, it is before every expiry there will ever be, and an +// implementation that passed it on would read every lease as expired, hand +// it to every instance, and restore the defect this lease fixes with a lease +// table sitting on top of it. +func dbNowMs(db *gorm.DB) (int64, error) { + expr, err := nowExprMs(db.Dialector.Name()) + if err != nil { + return 0, err + } + + var ms int64 + if err := db.Raw("SELECT " + expr).Row().Scan(&ms); err != nil { + return 0, fmt.Errorf("reading the database clock: %w", err) + } + if ms <= 0 { + return 0, fmt.Errorf("the database clock read as %d from %q", ms, expr) + } + return ms, nil +} + +// newOwnerID identifies this process in the lease row. +// +// Hostname and pid make a log line answer "which one is it" without a lookup; +// the random suffix is what actually makes it unique, because a container +// restarted under the same name can come back with the same hostname and the +// same pid 1. +func newOwnerID() string { + host, err := os.Hostname() + if err != nil || host == "" { + host = "unknown" + } + return fmt.Sprintf("%s-%d-%s", host, os.Getpid(), uuid.New().String()[:8]) +} + +// lease is one instance's claim on scheduling one database's jobs. +type lease struct { + db *gorm.DB + owner string + ttl time.Duration +} + +// acquire takes the lease or renews one this instance already holds, and +// reports whether this instance holds it when it returns. +// +// Renewal is tried first and is scoped to this owner, so it cannot take a +// lease another instance has meanwhile claimed. Only if that matches nothing +// does it try to take an expired one. Both are single UPDATE statements +// decided by RowsAffected: the database, not this process, arbitrates +// between two instances running this at the same moment. +// +// There is no insert path. The migration seeds the row, so a missing row is +// a broken installation rather than a state to recover from - and it is +// reported as one, instead of being papered over by an insert that two +// instances would race to win. +func (l *lease) acquire() (bool, error) { + nowMs, err := dbNowMs(l.db) + if err != nil { + return false, err + } + expiresMs := nowMs + l.ttl.Milliseconds() + + renewed := l.db.Model(&models2.SysJobLease{}). + Where("name = ? AND owner = ?", models2.SchedulerLeaseName, l.owner). + Update("expires_at_ms", expiresMs) + if renewed.Error != nil { + return false, fmt.Errorf("renewing the scheduler lease: %w", renewed.Error) + } + if renewed.RowsAffected > 0 { + return true, nil + } + + taken := l.db.Model(&models2.SysJobLease{}). + Where("name = ? AND expires_at_ms <= ?", models2.SchedulerLeaseName, nowMs). + Updates(map[string]any{ + "owner": l.owner, + "acquired_at_ms": nowMs, + "expires_at_ms": expiresMs, + }) + if taken.Error != nil { + return false, fmt.Errorf("taking the scheduler lease: %w", taken.Error) + } + if taken.RowsAffected > 0 { + return true, nil + } + + // Neither statement matched. Either another instance holds an + // unexpired lease - the ordinary case, and not an error - or the row + // the migration seeds is gone, which is, and which would otherwise + // present as jobs silently never running anywhere. + var rows int64 + if err := l.db.Model(&models2.SysJobLease{}). + Where("name = ?", models2.SchedulerLeaseName). + Count(&rows).Error; err != nil { + return false, fmt.Errorf("checking for the scheduler lease row: %w", err) + } + if rows == 0 { + return false, fmt.Errorf("the %q lease row is missing from %s; run the migrations", + models2.SchedulerLeaseName, (&models2.SysJobLease{}).TableName()) + } + return false, nil +} + +// release hands the lease back so a successor can take it now instead of +// waiting out the TTL. It is scoped to this owner: an instance that already +// lost the lease must not clear the row its successor is holding. +func (l *lease) release() error { + res := l.db.Model(&models2.SysJobLease{}). + Where("name = ? AND owner = ?", models2.SchedulerLeaseName, l.owner). + Updates(map[string]any{"owner": "", "expires_at_ms": 0}) + if res.Error != nil { + return fmt.Errorf("releasing the scheduler lease: %w", res.Error) + } + return nil +} diff --git a/app/jobs/lease_test.go b/app/jobs/lease_test.go new file mode 100644 index 00000000..e90770b4 --- /dev/null +++ b/app/jobs/lease_test.go @@ -0,0 +1,268 @@ +package jobs + +import ( + "os" + "testing" + "time" + + glebarez "github.com/glebarez/sqlite" + "gorm.io/driver/mysql" + "gorm.io/driver/postgres" + "gorm.io/driver/sqlserver" + "gorm.io/gorm" + + models2 "go-admin/app/jobs/models" +) + +// The lease is the one thing in this package whose correctness is a property +// of the database rather than of this process, so these run against every +// dialect that can be reached. SQLite always; the others when their DSN is +// set, and they must be set in CI - a suite that quietly skipped them would +// report success for a lease that cannot be taken at all on the dialect most +// installations actually run. +const ( + mysqlDSNEnv = "GO_ADMIN_TEST_MYSQL_DSN" + postgresDSNEnv = "GO_ADMIN_TEST_POSTGRES_DSN" + sqlserverDSNEnv = "GO_ADMIN_TEST_SQLSERVER_DSN" +) + +type dialectDB struct { + name string + open func(string) gorm.Dialector + env string +} + +var optionalDialects = []dialectDB{ + {"mysql", func(dsn string) gorm.Dialector { return mysql.Open(dsn) }, mysqlDSNEnv}, + {"postgres", func(dsn string) gorm.Dialector { return postgres.Open(dsn) }, postgresDSNEnv}, + {"sqlserver", func(dsn string) gorm.Dialector { return sqlserver.Open(dsn) }, sqlserverDSNEnv}, +} + +// eachDialect runs body against SQLite and against every optional dialect +// whose DSN is set. +func eachDialect(t *testing.T, body func(t *testing.T, db *gorm.DB)) { + t.Helper() + + t.Run("sqlite", func(t *testing.T) { + db, err := gorm.Open(glebarez.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatalf("opening sqlite: %v", err) + } + body(t, seedLeaseTable(t, db)) + }) + + for _, d := range optionalDialects { + t.Run(d.name, func(t *testing.T) { + dsn := os.Getenv(d.env) + if dsn == "" { + if os.Getenv("CI") != "" { + t.Fatalf("%s is not set while CI is: the lease must not go untested on %s", d.env, d.name) + } + t.Skipf("%s is not set; skipping %s", d.env, d.name) + } + db, err := gorm.Open(d.open(dsn), &gorm.Config{}) + if err != nil { + t.Fatalf("connecting to %s: %v", d.env, err) + } + sqlDB, err := db.DB() + if err != nil { + t.Fatalf("sql.DB: %v", err) + } + t.Cleanup(func() { _ = sqlDB.Close() }) + body(t, seedLeaseTable(t, db)) + }) + } +} + +// seedLeaseTable builds the shape 1786700009000 leaves behind: the table, +// and exactly one free row. +func seedLeaseTable(t *testing.T, db *gorm.DB) *gorm.DB { + t.Helper() + + if err := db.Migrator().DropTable(&models2.SysJobLease{}); err != nil { + t.Fatalf("dropping sys_job_lease: %v", err) + } + if err := db.AutoMigrate(&models2.SysJobLease{}); err != nil { + t.Fatalf("creating sys_job_lease: %v", err) + } + row := models2.SysJobLease{Name: models2.SchedulerLeaseName, AcquiredAtMs: 0, ExpiresAtMs: 0} + if err := db.Create(&row).Error; err != nil { + t.Fatalf("seeding the lease row: %v", err) + } + return db +} + +func TestTheDatabaseClockIsReadableAndIsNotTheZeroTime(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + nowMs, err := dbNowMs(db) + if err != nil { + t.Fatalf("dbNowMs: %v", err) + } + if nowMs <= 0 { + t.Fatal("the database clock read as zero, which would read every lease as expired") + } + // Not an assertion about either clock's accuracy - a container's + // clock and this one can drift. An hour is far wider than drift + // and far narrower than a timezone offset, which is the mistake + // this catches: reading MySQL's UTC_TIMESTAMP over a loc=Local + // DSN lands exactly one zone offset away and is invisible to + // every assertion that only compares the lease against itself. + drift := time.Duration(time.Now().UnixMilli()-nowMs) * time.Millisecond + if drift > time.Hour || drift < -time.Hour { + t.Errorf("the database clock is %v away from this process's; a timezone mistake looks exactly like this", drift) + } + }) +} + +func TestOnlyOneOfTwoInstancesTakesTheLease(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + a := &lease{db: db, owner: "instance-a", ttl: time.Minute} + b := &lease{db: db, owner: "instance-b", ttl: time.Minute} + + heldA, err := a.acquire() + if err != nil { + t.Fatalf("a.acquire: %v", err) + } + if !heldA { + t.Fatal("the first instance did not take a free lease") + } + + heldB, err := b.acquire() + if err != nil { + t.Fatalf("b.acquire: %v", err) + } + if heldB { + t.Error("the second instance took a lease the first one holds: both would schedule") + } + }) +} + +func TestTheHolderRenewsAndTheOtherStillCannotTakeIt(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + a := &lease{db: db, owner: "instance-a", ttl: time.Minute} + b := &lease{db: db, owner: "instance-b", ttl: time.Minute} + + if held, err := a.acquire(); err != nil || !held { + t.Fatalf("a.acquire: held=%v err=%v", held, err) + } + before := readLease(t, db) + + if held, err := a.acquire(); err != nil || !held { + t.Fatalf("a renewing: held=%v err=%v", held, err) + } + after := readLease(t, db) + + if after.ExpiresAtMs < before.ExpiresAtMs { + t.Errorf("renewal moved the expiry backwards: %d then %d", before.ExpiresAtMs, after.ExpiresAtMs) + } + if after.AcquiredAtMs != before.AcquiredAtMs { + t.Errorf("renewal moved acquired_at_ms (%d then %d); it must say when the lease was taken, not when it was last renewed", + before.AcquiredAtMs, after.AcquiredAtMs) + } + if held, err := b.acquire(); err != nil || held { + t.Errorf("the other instance took a renewed lease: held=%v err=%v", held, err) + } + }) +} + +func TestAnExpiredLeaseIsTakenOver(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + a := &lease{db: db, owner: "instance-a", ttl: time.Minute} + b := &lease{db: db, owner: "instance-b", ttl: time.Minute} + + if held, err := a.acquire(); err != nil || !held { + t.Fatalf("a.acquire: held=%v err=%v", held, err) + } + + // What a dead leader leaves behind: its row, unrenewed, past its + // expiry. Forced rather than waited out, so the test does not + // trade a second of sleep for the same assertion. + expire(t, db) + + if held, err := b.acquire(); err != nil || !held { + t.Fatalf("the successor did not take an expired lease: held=%v err=%v", held, err) + } + if got := readLease(t, db).Owner; got != "instance-b" { + t.Errorf("owner is %q after takeover, want instance-b", got) + } + + // And the instance that lost it must not get it back by renewing: + // renewal is scoped to the owner column it no longer matches. + if held, err := a.acquire(); err != nil || held { + t.Errorf("the dead leader renewed a lease it had lost: held=%v err=%v", held, err) + } + }) +} + +func TestReleaseHandsTheLeaseOnWithoutWaitingOutTheTTL(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + a := &lease{db: db, owner: "instance-a", ttl: time.Hour} + b := &lease{db: db, owner: "instance-b", ttl: time.Minute} + + if held, err := a.acquire(); err != nil || !held { + t.Fatalf("a.acquire: held=%v err=%v", held, err) + } + if held, err := b.acquire(); err != nil || held { + t.Fatalf("precondition: b must not hold it yet (held=%v err=%v)", held, err) + } + + if err := a.release(); err != nil { + t.Fatalf("a.release: %v", err) + } + if held, err := b.acquire(); err != nil || !held { + t.Errorf("a released a lease with an hour left and the successor still could not take it: held=%v err=%v", held, err) + } + }) +} + +func TestReleasingALeaseSomebodyElseHoldsDoesNothing(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + a := &lease{db: db, owner: "instance-a", ttl: time.Minute} + stale := &lease{db: db, owner: "instance-gone", ttl: time.Minute} + + if held, err := a.acquire(); err != nil || !held { + t.Fatalf("a.acquire: held=%v err=%v", held, err) + } + if err := stale.release(); err != nil { + t.Fatalf("stale.release: %v", err) + } + if got := readLease(t, db).Owner; got != "instance-a" { + t.Errorf("owner is %q; an instance that already lost the lease cleared its successor's row", got) + } + }) +} + +func TestAMissingLeaseRowIsReportedRatherThanSilentlyNeverScheduling(t *testing.T) { + eachDialect(t, func(t *testing.T, db *gorm.DB) { + if err := db.Where("name = ?", models2.SchedulerLeaseName). + Delete(&models2.SysJobLease{}).Error; err != nil { + t.Fatalf("deleting the lease row: %v", err) + } + a := &lease{db: db, owner: "instance-a", ttl: time.Minute} + held, err := a.acquire() + if held { + t.Fatal("acquire reported the lease held with no row to hold") + } + if err == nil { + t.Error("a missing lease row was reported as an ordinary 'someone else holds it': jobs would never run anywhere and nothing would say why") + } + }) +} + +func readLease(t *testing.T, db *gorm.DB) models2.SysJobLease { + t.Helper() + var row models2.SysJobLease + if err := db.Where("name = ?", models2.SchedulerLeaseName).First(&row).Error; err != nil { + t.Fatalf("reading the lease row: %v", err) + } + return row +} + +func expire(t *testing.T, db *gorm.DB) { + t.Helper() + if err := db.Model(&models2.SysJobLease{}). + Where("name = ?", models2.SchedulerLeaseName). + Update("expires_at_ms", 0).Error; err != nil { + t.Fatalf("expiring the lease: %v", err) + } +} diff --git a/app/jobs/models/sys_job_lease.go b/app/jobs/models/sys_job_lease.go new file mode 100644 index 00000000..b8cfd9ac --- /dev/null +++ b/app/jobs/models/sys_job_lease.go @@ -0,0 +1,58 @@ +package models + +// SchedulerLeaseName is the name of the one lease row per database. +// +// One row, not one per tenant: a tenant is a separate database with its own +// sys_job table and its own scheduler, so the row that decides who schedules +// it lives in that database alongside the jobs it governs. +const SchedulerLeaseName = "scheduler" + +// SysJobLease is the scheduler's single-writer lease over one database. +// +// app/jobs registers every enabled job into an in-process cron.Cron and keeps +// each job's scheduler handle in sys_job.entry_id. The scheduler is per +// process and entry_id is one shared column, so a second instance pointed at +// the same database does not divide the work - it overwrites it, and nothing +// logs that it did (issue #915). Only the holder of this lease calls +// jobs.Setup, which keeps the scheduler single-writer while the HTTP side +// still scales. +// +// It deliberately embeds neither models.ModelTime nor models.ControlBy. A +// lease is machine state, not a record a person creates, edits or +// soft-deletes: there is no author to attribute it to, and a deleted-but- +// present lease row would be a row that both does and does not hold the +// scheduler. +type SysJobLease struct { + // Name is the lease being held. The migration seeds exactly one row, + // SchedulerLeaseName, and the runtime only ever updates it - there is + // no insert path, so two instances starting at once cannot race to + // create the row they are both trying to claim. + Name string `json:"name" gorm:"type:varchar(64);primaryKey"` + + // Owner identifies the process that holds the lease. Empty means the + // lease is free, which is what the migration seeds. + Owner string `json:"owner" gorm:"type:varchar(191);not null"` + + // AcquiredAtMs is when the current owner took the lease, not when it + // last renewed: a leader that has held it for an hour and one that took + // over a second ago are different situations, and only this column + // tells them apart. Renewal moves ExpiresAtMs and leaves this alone. + AcquiredAtMs int64 `json:"acquiredAtMs" gorm:"column:acquired_at_ms;not null"` + + // ExpiresAtMs is when another instance may take the lease. + // + // Milliseconds since the Unix epoch, in a BIGINT, rather than a + // timestamp column. A timestamp crossing the driver boundary carries + // timezone semantics that the driver applies on the way through: with + // go-admin's own `parseTime=True&loc=Local` DSN, MySQL's UTC_TIMESTAMP + // comes back labelled as local time, and a lease written in Asia/ + // Shanghai is then eight hours out - in whichever direction makes every + // other instance's lease look expired. An integer has no timezone for + // anything to apply, and the comparison that decides who schedules + // becomes integer arithmetic that no DSN setting can reinterpret. + ExpiresAtMs int64 `json:"expiresAtMs" gorm:"column:expires_at_ms;not null"` +} + +func (*SysJobLease) TableName() string { + return "sys_job_lease" +} diff --git a/app/jobs/supervisor.go b/app/jobs/supervisor.go new file mode 100644 index 00000000..34481efa --- /dev/null +++ b/app/jobs/supervisor.go @@ -0,0 +1,212 @@ +package jobs + +import ( + "context" + "sync" + "time" + + log "github.com/go-admin-team/go-admin-core/v2/logger" + "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/pkg/cronjob" + "gorm.io/gorm" +) + +// leaseTTL is how long a lease stays valid without being renewed, and +// leaseHeartbeat is how often the holder renews it. +// +// The gap between them is the point: at a third of the TTL, two consecutive +// renewals can fail - a restarting database, a paused container - and the +// third still lands before anything else may take the lease. Making them +// equal would hand the scheduler to another instance on the first missed +// beat. +// +// The TTL is also the longest the jobs can be stopped everywhere: an +// instance killed without running its shutdown leaves its lease behind, and +// the successor waits this long before taking it. +const ( + leaseTTL = 30 * time.Second + leaseHeartbeat = 10 * time.Second +) + +// supervisor keeps one tenant's scheduler in step with one lease. +// +// It exists because holding the lease is not a decision made once at +// startup. An instance that never gets the lease has to keep asking, or the +// death of the current holder would stop the jobs until somebody restarted a +// process by hand; and an instance that holds it has to stop scheduling the +// moment it can no longer prove it still does, or a network partition turns +// into the two-schedulers-at-once defect (#915) that the lease exists to +// prevent. +type supervisor struct { + key string + db *gorm.DB + lease *lease + + mu sync.Mutex + running bool + // lastRenew is when this instance last proved it holds the lease. It + // is compared only against this process's own later readings, never + // against another instance's, so the monotonic clock is the right one + // here - the reason the lease itself reads the database's clock does + // not apply to measuring how long ago something happened locally. + lastRenew time.Time + + stop chan struct{} + stopOnce sync.Once + done chan struct{} +} + +func newSupervisor(key string, db *gorm.DB, owner string) *supervisor { + return &supervisor{ + key: key, + db: db, + lease: &lease{db: db, owner: owner, ttl: leaseTTL}, + stop: make(chan struct{}), + done: make(chan struct{}), + } +} + +// start takes the lease if it is free, schedules this tenant's jobs if it +// got it, and then keeps both facts true for the life of the process. +// +// The first attempt is synchronous so that a single-instance deployment - +// which is nearly all of them - has its jobs registered by the time Setup +// returns, exactly as it did before there was a lease. +func (s *supervisor) start() { + s.tick() + + go s.heartbeat() + + sdk.Runtime.SetShutdown(func(ctx context.Context) { + s.shutdown(ctx) + }) +} + +func (s *supervisor) heartbeat() { + defer close(s.done) + + t := time.NewTicker(leaseHeartbeat) + defer t.Stop() + + for { + select { + case <-s.stop: + return + case <-t.C: + s.tick() + } + } +} + +// tick asks for the lease and makes the scheduler match the answer. +func (s *supervisor) tick() { + held, err := s.lease.acquire() + if err != nil { + // Not knowing is not the same as having lost it. The lease is + // still ours until it expires, so the scheduler keeps running + // and this instance keeps trying - a database that is briefly + // unreachable must not stop the jobs, and must not hand them to + // anyone else either, because nobody else can reach it to take + // the lease. + log.Errorf("[Job] scheduler lease for %s: %v", s.key, err) + if s.heldFor() > leaseTTL { + log.Errorf("[Job] scheduler lease for %s has not been renewed in %v; stopping the scheduler before anything else takes it", + s.key, leaseTTL) + s.stopScheduling() + } + return + } + + if !held { + s.stopScheduling() + return + } + + s.mu.Lock() + s.lastRenew = time.Now() + already := s.running + s.mu.Unlock() + + if !already { + s.startScheduling() + } +} + +// heldFor reports how long it has been since this instance last proved it +// holds the lease. A zero lastRenew means it never has, which is not a lease +// that has gone stale. +func (s *supervisor) heldFor() time.Duration { + s.mu.Lock() + defer s.mu.Unlock() + if s.lastRenew.IsZero() { + return 0 + } + return time.Since(s.lastRenew) +} + +func (s *supervisor) startScheduling() { + s.mu.Lock() + if s.running { + s.mu.Unlock() + return + } + s.running = true + s.mu.Unlock() + + log.Infof("[Job] holding the scheduler lease for %s; registering its jobs", s.key) + setup(s.key, s.db) +} + +// stopScheduling stops this tenant's scheduler and puts a fresh one in its +// place. +// +// Fresh, rather than reusing the stopped one, because taking the lease back +// runs setup again and setup adds every enabled job to whatever scheduler is +// registered. Reusing it would leave the previous registration in place and +// fire every job twice - the symptom this whole change is here to remove, +// reintroduced one layer down. +func (s *supervisor) stopScheduling() { + s.mu.Lock() + if !s.running { + s.mu.Unlock() + return + } + s.running = false + s.mu.Unlock() + + log.Infof("[Job] no longer holding the scheduler lease for %s; stopping its jobs", s.key) + ctx, cancel := context.WithTimeout(context.Background(), leaseHeartbeat) + defer cancel() + stopCrontab(ctx, s.key) + sdk.Runtime.SetCrontabByTenant(s.key, cronjob.NewWithSeconds()) +} + +// shutdown stops the heartbeat, stops the scheduler and hands the lease back +// so a successor can take it now rather than waiting out the TTL. +func (s *supervisor) shutdown(ctx context.Context) { + s.stopOnce.Do(func() { close(s.stop) }) + select { + case <-s.done: + case <-ctx.Done(): + } + + s.mu.Lock() + wasRunning := s.running + s.running = false + s.mu.Unlock() + + if wasRunning { + stopCrontab(ctx, s.key) + if err := s.lease.release(); err != nil { + log.Errorf("[Job] releasing the scheduler lease for %s: %v", s.key, err) + } + } +} + +// holdsLease reports whether this instance is currently scheduling. It exists +// for the tests: everything else acts on the answer inside tick. +func (s *supervisor) holdsLease() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.running +} diff --git a/app/jobs/supervisor_test.go b/app/jobs/supervisor_test.go new file mode 100644 index 00000000..9bf6466b --- /dev/null +++ b/app/jobs/supervisor_test.go @@ -0,0 +1,94 @@ +package jobs + +import ( + "testing" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/pkg/cronjob" + + models2 "go-admin/app/jobs/models" +) + +// An instance that starts while another one holds the lease must not +// register the jobs. This is the whole point: every instance registering the +// whole enabled list into its own scheduler is what made one job fire once +// per instance (#915). +func TestASecondInstanceDoesNotScheduleWhileTheFirstHoldsTheLease(t *testing.T) { + const tenant = "second-instance" + db := leaseDB(t) + sdk.Runtime.SetCrontabByTenant(tenant, cronjob.NewWithSeconds()) + + first := newSupervisor(tenant, db, "instance-a") + first.tick() + if !first.holdsLease() { + t.Fatal("the first instance did not take a free lease") + } + + second := newSupervisor(tenant, db, "instance-b") + second.tick() + if second.holdsLease() { + t.Error("a second instance scheduled while the first holds the lease: the job would fire twice per tick") + } +} + +// Losing the lease has to stop the scheduler, not merely stop it from being +// taken again. A holder that keeps scheduling after its lease has gone to +// somebody else is two schedulers at once - the defect the lease exists to +// prevent, reached from the other direction. +func TestTheSupervisorStopsSchedulingWhenItLosesTheLease(t *testing.T) { + const tenant = "loses-lease" + db := leaseDB(t) + sdk.Runtime.SetCrontabByTenant(tenant, cronjob.NewWithSeconds()) + + holder := newSupervisor(tenant, db, "instance-a") + holder.tick() + if !holder.holdsLease() { + t.Fatal("the supervisor did not take a free lease, so losing it cannot be observed") + } + + // What a partition looks like from the database's side: the lease + // lapsed and somebody else took it while this instance was away. + expire(t, db) + successor := &lease{db: db, owner: "instance-b", ttl: time.Minute} + if held, err := successor.acquire(); err != nil || !held { + t.Fatalf("the successor could not take the expired lease: held=%v err=%v", held, err) + } + + holder.tick() + + if holder.holdsLease() { + t.Error("the supervisor kept scheduling after the lease went to another instance") + } + if got := readLease(t, db).Owner; got != "instance-b" { + t.Errorf("owner is %q; the instance that lost the lease wrote over its successor", got) + } +} + +// A database that cannot be reached is not the same as a lease that has been +// lost. Stopping on the first failed renewal would stop the jobs every time +// the database blinked - and hand them to nobody, because no other instance +// can reach it to take the lease either. +func TestABrieflyUnreachableDatabaseDoesNotStopTheScheduler(t *testing.T) { + const tenant = "db-blip" + db := leaseDB(t) + sdk.Runtime.SetCrontabByTenant(tenant, cronjob.NewWithSeconds()) + + s := newSupervisor(tenant, db, "instance-a") + s.tick() + if !s.holdsLease() { + t.Fatal("the supervisor did not take a free lease") + } + + // The table going missing is how an unreachable database presents to + // acquire: every statement against it returns an error. + if err := db.Migrator().DropTable(&models2.SysJobLease{}); err != nil { + t.Fatalf("dropping the lease table: %v", err) + } + + s.tick() + + if !s.holdsLease() { + t.Error("one failed renewal stopped the scheduler; the lease had not expired yet and nobody else could have taken it") + } +} diff --git a/cmd/migrate/migration/version/1786700009000_job_scheduler_lease.go b/cmd/migrate/migration/version/1786700009000_job_scheduler_lease.go new file mode 100644 index 00000000..24669f82 --- /dev/null +++ b/cmd/migrate/migration/version/1786700009000_job_scheduler_lease.go @@ -0,0 +1,61 @@ +package version + +import ( + "runtime" + + "gorm.io/gorm" + + jobmodels "go-admin/app/jobs/models" + "go-admin/cmd/migrate/migration" + common "go-admin/common/models" +) + +// Create sys_job_lease and seed the one row the scheduler competes for +// (issue #915). +// +// The row is seeded here rather than created on demand at startup. Two +// instances starting together would otherwise race to insert the very row +// they are each trying to claim, and the loser would have to tell a +// duplicate-key error apart from a real one in whichever driver it is +// running against. Seeding it makes the runtime path two UPDATE statements +// and nothing else. +// +// It is seeded free - no owner, and an expiry far enough in the past that +// the first instance to ask takes it - so that installing this migration +// does not leave the scheduler waiting out a TTL that nobody is holding. +// +// Ordered after 1786700003000 (the soft-delete conversion), so importing +// cmd/migrate/migration/models is banned here - see +// schema_coverage_test.go's TestPostConversionMigrationsAvoidFrozenSeedModels. +// sys_job_lease is AutoMigrate'd from its runtime model under +// app/jobs/models directly, and it is absent from 1786700003000's frozen +// softDeleteTables list because it embeds no common.ModelTime: a lease that +// could be soft-deleted would be a row that both does and does not hold the +// scheduler. +func init() { + _, fileName, _, _ := runtime.Caller(0) + migration.Migrate.SetVersion(migration.GetFilename(fileName), _1786700009000JobSchedulerLease) +} + +func _1786700009000JobSchedulerLease(db *gorm.DB, version string) error { + return db.Transaction(func(tx *gorm.DB) error { + if err := tx.Migrator().AutoMigrate(new(jobmodels.SysJobLease)); err != nil { + return err + } + + // Seeded free: no owner, and an expiry of 0 - before every clock + // reading there will ever be - so the first instance to ask takes + // it rather than waiting out a TTL nobody is holding. + lease := jobmodels.SysJobLease{ + Name: jobmodels.SchedulerLeaseName, + Owner: "", + AcquiredAtMs: 0, + ExpiresAtMs: 0, + } + if err := tx.Create(&lease).Error; err != nil { + return err + } + + return tx.Create(&common.Migration{Version: version}).Error + }) +} diff --git a/cmd/migrate/migration/version/1786700009000_job_scheduler_lease_test.go b/cmd/migrate/migration/version/1786700009000_job_scheduler_lease_test.go new file mode 100644 index 00000000..eea0aed3 --- /dev/null +++ b/cmd/migrate/migration/version/1786700009000_job_scheduler_lease_test.go @@ -0,0 +1,75 @@ +package version + +import ( + "testing" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + + jobmodels "go-admin/app/jobs/models" + common "go-admin/common/models" +) + +func openJobLeaseDB(t *testing.T) *gorm.DB { + t.Helper() + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatalf("open: %v", err) + } + if err := db.AutoMigrate(&common.Migration{}); err != nil { + t.Fatalf("automigrate: %v", err) + } + return db +} + +// The runtime has no insert path - two instances starting together would +// race to create the row they are both trying to claim - so the row has to +// exist when the migration finishes or nothing ever schedules anything. +func TestTheSchedulerLeaseMigrationLeavesExactlyOneFreeRow(t *testing.T) { + db := openJobLeaseDB(t) + + if err := _1786700009000JobSchedulerLease(db, "1786700009000"); err != nil { + t.Fatalf("migrate: %v", err) + } + + if !db.Migrator().HasTable(&jobmodels.SysJobLease{}) { + t.Fatal("sys_job_lease was not created") + } + + var rows []jobmodels.SysJobLease + if err := db.Find(&rows).Error; err != nil { + t.Fatalf("reading sys_job_lease: %v", err) + } + if len(rows) != 1 { + t.Fatalf("sys_job_lease holds %d rows, want exactly 1", len(rows)) + } + + row := rows[0] + if row.Name != jobmodels.SchedulerLeaseName { + t.Errorf("the seeded row is named %q, want %q; acquire looks the row up by this name and would find nothing", + row.Name, jobmodels.SchedulerLeaseName) + } + if row.Owner != "" { + t.Errorf("the seeded lease is owned by %q; a fresh install would wait out a TTL held by nobody", row.Owner) + } + // Zero, not "now": the take is `expires_at_ms <= now`, so a seeded + // expiry in the future is a scheduler that does not start until it + // passes. + if row.ExpiresAtMs != 0 { + t.Errorf("the seeded lease expires at %d, want 0", row.ExpiresAtMs) + } +} + +func TestTheSchedulerLeaseMigrationRecordsItsVersion(t *testing.T) { + db := openJobLeaseDB(t) + + if err := _1786700009000JobSchedulerLease(db, "1786700009000"); err != nil { + t.Fatalf("migrate: %v", err) + } + + var got common.Migration + if err := db.Where("version = ?", "1786700009000").First(&got).Error; err != nil { + t.Fatalf("the migration did not record its version, so it would run again on every start: %v", err) + } +} diff --git a/scripts/k8s/deploy.yml b/scripts/k8s/deploy.yml index e9ad5e24..b408fc7e 100644 --- a/scripts/k8s/deploy.yml +++ b/scripts/k8s/deploy.yml @@ -23,20 +23,18 @@ metadata: version: v1 spec: # One replica, and the drain window below buys nothing at one replica: there - # is nowhere to send the traffic this pod stops taking. Raising it needs two - # changes that are not this number: + # is nowhere to send the traffic this pod stops taking. # - # The volume below is shared by every replica, and the log path in - # settings.yml lives on it, so a second pod would append to the same - # rotating file. + # The scheduler no longer stands in the way of raising this. Every pod takes + # a lease row in its own database (sys_job_lease) and only the holder + # registers the jobs, so one enabled job fires once however many pods there + # are; a pod that loses the lease stops scheduling, and one that exits hands + # it back so a successor starts without waiting out the lease. See #915. # - # The job scheduler is per process while its handle on a job is one shared - # column. Startup runs `UPDATE sys_job SET entry_id = 0 WHERE entry_id > 0` - # across the whole table (app/jobs/jobbase.go), so a second pod erases the - # first pod's ids and writes its own, and every pod registers the whole - # enabled list in its own scheduler. Neither symptom logs anything: an - # enabled job fires once per pod, and stopping one from the UI removes an - # entry from the wrong process and still answers 200. See #915. + # What still does stand in the way: the volume below is shared by every + # replica, and the log path in settings.yml lives on it, so a second pod + # appends to the same rotating file. Give each replica its own log + # destination before raising this. replicas: 1 selector: matchLabels: