《后端架构设计·分布式与数据一致性》11.1 调度模型：轮询、租约与领导者选举

// file: internal/platform/scheduler/lease.go
// tier: L2
// verify: go test ./internal/platform/scheduler/ -count=1
package scheduler

import (
	"errors"
	"fmt"
	"sync"
)

var (
	// ErrLeaseHeld means another owner still holds an unexpired lease.
	ErrLeaseHeld = errors.New("scheduler: lease held")
	// ErrFenced means the caller acted with a superseded epoch.
	ErrFenced = errors.New("scheduler: stale epoch")
)

// ManualClock is an injected tick clock, so tests never sleep.
type ManualClock struct {
	mu  sync.Mutex
	now int64
}

// Now returns the current tick.
func (c *ManualClock) Now() int64 {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.now
}

// Advance moves the clock forward by d ticks.
func (c *ManualClock) Advance(d int64) {
	c.mu.Lock()
	c.now += d
	c.mu.Unlock()
}

// Lease is the right to dispatch for one term. Epoch is a fencing token that
// increases on every acquisition and never decreases.
type Lease struct {
	Key       string
	Owner     string
	Epoch     uint64
	ExpiresAt int64
}

// LeaseStore grants at most one unexpired lease per key. It stands in for a
// row updated by compare-and-set: (owner, epoch, expires_at).
type LeaseStore struct {
	mu    sync.Mutex
	now   func() int64
	terms map[string]Lease
}

// NewLeaseStore builds a store on top of now.
func NewLeaseStore(now func() int64) *LeaseStore {
	return &LeaseStore{now: now, terms: make(map[string]Lease)}
}

// Acquire takes the lease for owner. It fails while another owner's lease is
// unexpired; an expired lease is free and the new term gets a higher epoch.
func (s *LeaseStore) Acquire(key, owner string, ttl int64) (Lease, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := s.now()
	cur, ok := s.terms[key]
	if ok && cur.Owner != owner && now < cur.ExpiresAt {
		return Lease{}, fmt.Errorf("acquire %s for %s: %w", key, owner, ErrLeaseHeld)
	}
	var epoch uint64 = 1
	if ok {
		epoch = cur.Epoch + 1
	}
	l := Lease{Key: key, Owner: owner, Epoch: epoch, ExpiresAt: now + ttl}
	s.terms[key] = l
	return l, nil
}

// Renew extends a lease only for its current owner and epoch, and only before
// expiry. A paused owner that comes back after takeover cannot renew.
func (s *LeaseStore) Renew(l Lease, ttl int64) (Lease, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	cur, ok := s.terms[l.Key]
	if !ok || cur.Owner != l.Owner || cur.Epoch != l.Epoch || s.now() >= cur.ExpiresAt {
		return Lease{}, fmt.Errorf("renew %s epoch %d: %w", l.Key, l.Epoch, ErrFenced)
	}
	l.ExpiresAt = s.now() + ttl
	s.terms[l.Key] = l
	return l, nil
}

// Current returns the epoch most recently granted for key, or zero.
func (s *LeaseStore) Current(key string) uint64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.terms[key].Epoch
}

// Sink writes orders only when the writer carries the current epoch. It reads
// the authoritative term instead of trusting the caller.
type Sink struct {
	mu      sync.Mutex
	current func() uint64
	applied int
}

// NewSink wires a sink to the store that grants dispatch terms.
func NewSink(store *LeaseStore) *Sink {
	return &Sink{current: func() uint64 { return store.Current("orders:dispatch") }}
}

// Write applies orderID only if l carries the current epoch.
func (s *Sink) Write(l Lease, orderID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if l.Epoch < s.current() {
		return fmt.Errorf("write %s epoch %d: %w", orderID, l.Epoch, ErrFenced)
	}
	s.applied++
	return nil
}

// Len reports how many writes were applied.
func (s *Sink) Len() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.applied
}

// Elector is a non-blocking campaign loop driven by Tick. A crashed process
// simply stops calling Tick, so its lease expires.
type Elector struct {
	store *LeaseStore
	owner string
	ttl   int64
	lease Lease
	lead  bool
}

// NewElector creates an elector for owner with the given lease TTL.
func NewElector(store *LeaseStore, owner string, ttl int64) *Elector {
	return &Elector{store: store, owner: owner, ttl: ttl}
}

// Tick renews leadership or campaigns for it.
func (e *Elector) Tick() (Lease, bool, error) {
	if e.lead {
		l, err := e.store.Renew(e.lease, e.ttl)
		if err == nil {
			e.lease = l
			return l, true, nil
		}
		e.lead = false
	}
	l, err := e.store.Acquire("orders:dispatch", e.owner, e.ttl)
	if err != nil {
		return Lease{}, false, err
	}
	e.lease, e.lead = l, true
	return l, true, nil
}

=== RUN   TestSimulateModes
    lease_test.go:103: scan effects=60 duplicates=40 empty_polls=0 failover=0
    lease_test.go:103: claim effects=23 duplicates=3 empty_polls=12 failover=0
    lease_test.go:103: leader effects=20 duplicates=0 empty_polls=1 failover=7
--- PASS: TestSimulateModes (0.00s)