Merge pull request #947 from go-admin-team/fix/915-scheduler-lease

fix🐛: make the job scheduler single-writer with a database lease

.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:
 

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)

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
 }

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
+}

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)
+	}
+}

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"
+}

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
+}

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")
+	}
+}

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
+	})
+}

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)
+	}
+}

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: