release: v0.0.13 — scheduler leader election (HA, root-fix duplicate billing)

Lease-based leader election (internal/leader) runs the singleton billing/auto-recharge/alert scheduler on exactly one replica; scheduler made re-startable with tests; replicaCount restored to 2; LEADER_ELECTION_ENABLED toggle; coordination.k8s.io/leases RBAC added.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: SuperMarioYL

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to the Bison project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.0.13] - 2026-06-19
+
+### Added — Scheduler leader election
+
+- **Lease-based leader election** (`internal/leader`) guards the singleton billing/auto-recharge/alert scheduler so it runs on exactly one api-server replica at a time. This is the root fix for the duplicate-billing risk; `apiServer.replicaCount` is restored to `2` for HA.
+- The scheduler is now **re-startable** (clean `Start`/`Stop` on leadership changes), with new tests covering restart and stop-before-start safety.
+- Toggle via `LEADER_ELECTION_ENABLED` (default on); disable for single-replica / local dev.
+- Added `coordination.k8s.io/leases` (get/create/update) to the api-server RBAC.
+
 ## [0.0.12] - 2026-06-19
 
 ### Fixed — Billing correctness & concurrency

api-server/cmd/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/bison/api-server/internal/config"
 	"github.com/bison/api-server/internal/handler"
 	"github.com/bison/api-server/internal/k8s"
+	"github.com/bison/api-server/internal/leader"
 	"github.com/bison/api-server/internal/middleware"
 	"github.com/bison/api-server/internal/opencost"
 	"github.com/bison/api-server/internal/scheduler"
@@ -300,9 +301,22 @@ func main() {
 		IdleTimeout:  60 * time.Second,
 	}
 
-	// Start scheduler
+	// Start scheduler. When leader election is enabled the singleton scheduler
+	// runs on exactly one replica at a time (guards against duplicate billing);
+	// otherwise it runs directly (single-replica / local dev).
 	ctx, cancel := context.WithCancel(context.Background())
-	sched.Start(ctx)
+	if cfg.LeaderElectionEnabled {
+		go leader.Run(ctx, k8sClient.Clientset(), service.BisonNamespace,
+			func(leaderCtx context.Context) {
+				sched.Start(leaderCtx)
+				<-leaderCtx.Done()
+				sched.Stop()
+			},
+			func() { sched.Stop() },
+		)
+	} else {
+		sched.Start(ctx)
+	}
 
 	// Start server in goroutine
 	go func() {

api-server/internal/config/config.go

@@ -24,6 +24,11 @@ type Config struct {
 
 	// Feature toggles
 	CapsuleEnabled bool
+
+	// LeaderElectionEnabled gates the singleton scheduler behind a Kubernetes
+	// lease so it runs on exactly one replica. Disable for single-replica or
+	// out-of-cluster development.
+	LeaderElectionEnabled bool
 }
 
 // Load reads configuration from environment variables
@@ -35,9 +40,10 @@ func Load() (*Config, error) {
 		AdminUsername: "admin",
 		AdminPassword: "admin",
 		JWTSecret:     "bison-secret-key-change-in-production",
-		OpenCostURL:    "",
-		PrometheusURL:  "",
-		CapsuleEnabled: true,
+		OpenCostURL:           "",
+		PrometheusURL:         "",
+		CapsuleEnabled:        true,
+		LeaderElectionEnabled: true,
 	}
 
 	if port := os.Getenv("PORT"); port != "" {
@@ -78,6 +84,9 @@ func Load() (*Config, error) {
 	if capsuleEnabled := os.Getenv("CAPSULE_ENABLED"); capsuleEnabled == "false" {
 		cfg.CapsuleEnabled = false
 	}
+	if le := os.Getenv("LEADER_ELECTION_ENABLED"); le == "false" {
+		cfg.LeaderElectionEnabled = false
+	}
 
 	return cfg, nil
 }

api-server/internal/k8s/client.go

@@ -37,6 +37,11 @@ func NewClientWithInterfaces(clientset kubernetes.Interface, dynamicClient dynam
 	}
 }
 
+// Clientset exposes the underlying typed client (used e.g. for leader election).
+func (c *Client) Clientset() kubernetes.Interface {
+	return c.clientset
+}
+
 // NewClient creates a new Kubernetes client
 func NewClient() (*Client, error) {
 	var config *rest.Config

api-server/internal/leader/leader.go

@@ -0,0 +1,98 @@
+// Package leader provides Kubernetes lease-based leader election so that
+// singleton background work (the billing/auto-recharge/alert scheduler) runs on
+// exactly one api-server replica at a time, even when scaled horizontally.
+package leader
+
+import (
+	"context"
+	"os"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+
+	"github.com/bison/api-server/pkg/logger"
+)
+
+// LeaseName is the coordination.k8s.io Lease used to elect the scheduler leader.
+const LeaseName = "bison-scheduler"
+
+// identity returns a per-process identity. In Kubernetes the pod name is the
+// hostname, which is unique per replica; POD_NAME overrides it when set.
+func identity() string {
+	if v := os.Getenv("POD_NAME"); v != "" {
+		return v
+	}
+	if h, err := os.Hostname(); err == nil && h != "" {
+		return h
+	}
+	return "bison-api"
+}
+
+func namespace(def string) string {
+	if v := os.Getenv("POD_NAMESPACE"); v != "" {
+		return v
+	}
+	return def
+}
+
+// Run blocks running leader election until ctx is cancelled.
+//
+// onStarted is invoked (in its own goroutine) with a context that is cancelled
+// when leadership is lost or ctx is cancelled; it should start the leader-only
+// work and return promptly when its context is done. onStopped is invoked when
+// leadership is lost. The scheduler must be re-startable, since leadership can be
+// re-acquired after a transient loss.
+func Run(ctx context.Context, clientset kubernetes.Interface, ns string, onStarted func(context.Context), onStopped func()) {
+	id := identity()
+	leaseNS := namespace(ns)
+
+	lock := &resourcelock.LeaseLock{
+		LeaseMeta:  metav1.ObjectMeta{Name: LeaseName, Namespace: leaseNS},
+		Client:     clientset.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{Identity: id},
+	}
+
+	logger.Info("Starting leader election", "identity", id, "namespace", leaseNS, "lease", LeaseName)
+
+	config := leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		ReleaseOnCancel: true,
+		LeaseDuration:   15 * time.Second,
+		RenewDeadline:   10 * time.Second,
+		RetryPeriod:     2 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(leaderCtx context.Context) {
+				logger.Info("Acquired scheduler leadership", "identity", id)
+				onStarted(leaderCtx)
+			},
+			OnStoppedLeading: func() {
+				logger.Warn("Lost scheduler leadership", "identity", id)
+				onStopped()
+			},
+			OnNewLeader: func(current string) {
+				if current != id {
+					logger.Info("Observed scheduler leader", "leader", current)
+				}
+			},
+		},
+	}
+
+	// RunOrDie returns when ctx is cancelled or leadership is lost. Loop so a
+	// transient loss leads to re-election rather than permanently idle.
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		leaderelection.RunOrDie(ctx, config)
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(2 * time.Second):
+		}
+	}
+}

api-server/internal/scheduler/scheduler.go

@@ -19,8 +19,10 @@ type Scheduler struct {
 	executions   []service.TaskExecution
 	executionsMu sync.RWMutex
 
-	stopCh chan struct{}
-	wg     sync.WaitGroup
+	mu      sync.Mutex
+	started bool
+	stopCh  chan struct{}
+	wg      sync.WaitGroup
 }
 
 // NewScheduler creates a new Scheduler
@@ -34,31 +36,41 @@ func NewScheduler(
 		balanceSvc: balanceSvc,
 		alertSvc:   alertSvc,
 		executions: make([]service.TaskExecution, 0),
-		stopCh:     make(chan struct{}),
 	}
 }
 
-// Start starts all scheduled tasks
+// Start starts all scheduled tasks. It is idempotent and re-startable: calling
+// Start after a Stop (e.g. when leadership is re-acquired) spins up a fresh set
+// of tasks against a new stop channel.
 func (s *Scheduler) Start(ctx context.Context) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.started {
+		return
+	}
+	s.started = true
+	s.stopCh = make(chan struct{})
 	logger.Info("Starting scheduler")
 
-	// Start billing task (every hour)
-	s.wg.Add(1)
+	s.wg.Add(3)
 	go s.runBillingTask(ctx)
-
-	// Start auto-recharge task (every hour)
-	s.wg.Add(1)
 	go s.runAutoRechargeTask(ctx)
-
-	// Start alert check task (every 15 minutes)
-	s.wg.Add(1)
 	go s.runAlertTask(ctx)
 }
 
-// Stop stops all scheduled tasks
+// Stop stops all scheduled tasks and waits for them to exit. Safe to call when
+// not started.
 func (s *Scheduler) Stop() {
-	logger.Info("Stopping scheduler")
+	s.mu.Lock()
+	if !s.started {
+		s.mu.Unlock()
+		return
+	}
+	s.started = false
 	close(s.stopCh)
+	s.mu.Unlock()
+
+	logger.Info("Stopping scheduler")
 	s.wg.Wait()
 }
 
@@ -101,14 +113,14 @@ func (s *Scheduler) safeExecute(name string, fn func()) {
 
 // sleepWithJitter waits a random duration in [0, max) to desynchronize task
 // firing across replicas, returning false if the scheduler is stopped meanwhile.
-func (s *Scheduler) sleepWithJitter(max time.Duration) bool {
+func (s *Scheduler) sleepWithJitter(stopCh <-chan struct{}, max time.Duration) bool {
 	if max <= 0 {
 		return true
 	}
 	timer := time.NewTimer(time.Duration(rand.Int63n(int64(max))))
 	defer timer.Stop()
 	select {
-	case <-s.stopCh:
+	case <-stopCh:
 		return false
 	case <-timer.C:
 		return true
@@ -117,8 +129,9 @@ func (s *Scheduler) sleepWithJitter(max time.Duration) bool {
 
 func (s *Scheduler) runBillingTask(ctx context.Context) {
 	defer s.wg.Done()
+	stopCh := s.stopCh
 
-	if !s.sleepWithJitter(60 * time.Second) {
+	if !s.sleepWithJitter(stopCh, 60*time.Second) {
 		return
 	}
 
@@ -127,7 +140,7 @@ func (s *Scheduler) runBillingTask(ctx context.Context) {
 
 	for {
 		select {
-		case <-s.stopCh:
+		case <-stopCh:
 			return
 		case <-ticker.C:
 			s.safeExecute("billing", func() { s.executeBillingTask(ctx) })
@@ -161,8 +174,9 @@ func (s *Scheduler) executeBillingTask(ctx context.Context) {
 
 func (s *Scheduler) runAutoRechargeTask(ctx context.Context) {
 	defer s.wg.Done()
+	stopCh := s.stopCh
 
-	if !s.sleepWithJitter(60 * time.Second) {
+	if !s.sleepWithJitter(stopCh, 60*time.Second) {
 		return
 	}
 
@@ -171,7 +185,7 @@ func (s *Scheduler) runAutoRechargeTask(ctx context.Context) {
 
 	for {
 		select {
-		case <-s.stopCh:
+		case <-stopCh:
 			return
 		case <-ticker.C:
 			s.safeExecute("auto_recharge", func() { s.executeAutoRechargeTask(ctx) })
@@ -205,8 +219,9 @@ func (s *Scheduler) executeAutoRechargeTask(ctx context.Context) {
 
 func (s *Scheduler) runAlertTask(ctx context.Context) {
 	defer s.wg.Done()
+	stopCh := s.stopCh
 
-	if !s.sleepWithJitter(30 * time.Second) {
+	if !s.sleepWithJitter(stopCh, 30*time.Second) {
 		return
 	}
 
@@ -215,7 +230,7 @@ func (s *Scheduler) runAlertTask(ctx context.Context) {
 
 	for {
 		select {
-		case <-s.stopCh:
+		case <-stopCh:
 			return
 		case <-ticker.C:
 			s.safeExecute("alert_check", func() { s.executeAlertTask(ctx) })

api-server/internal/scheduler/scheduler_test.go

@@ -0,0 +1,45 @@
+package scheduler
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/bison/api-server/pkg/logger"
+)
+
+func TestMain(m *testing.M) {
+	logger.Init(false)
+	os.Exit(m.Run())
+}
+
+// TestSchedulerRestartable verifies the scheduler can be stopped and started
+// again (required for leader-election re-acquisition) and that Start/Stop are
+// idempotent and do not deadlock.
+func TestSchedulerRestartable(t *testing.T) {
+	s := NewScheduler(nil, nil, nil)
+	ctx := context.Background()
+
+	done := make(chan struct{})
+	go func() {
+		s.Start(ctx)
+		s.Start(ctx) // idempotent: second Start is a no-op
+		s.Stop()
+		s.Stop()     // idempotent: second Stop is a no-op
+		s.Start(ctx) // re-startable after Stop
+		s.Stop()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Start/Stop deadlocked")
+	}
+}
+
+func TestStopBeforeStartIsSafe(t *testing.T) {
+	s := NewScheduler(nil, nil, nil)
+	s.Stop() // must not panic or block when never started
+}

deploy/charts/bison/Chart.yaml

@@ -2,8 +2,8 @@ apiVersion: v2
 name: bison
 description: Bison - GPU 资源计费平台，基于 Capsule 多租户 + OpenCost 成本追踪
 type: application
-version: 0.0.12
-appVersion: "0.0.12"
+version: 0.0.13
+appVersion: "0.0.13"
 keywords:
   - gpu
   - billing