《后端架构设计·分布式与数据一致性》11.2 任务分片、幂等重跑与进度检查点

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

import (
	"errors"
	"fmt"
	"hash/fnv"
	"sync"
)

var (
	// ErrStaleEpoch means a worker from a superseded shard term tried to write.
	ErrStaleEpoch = errors.New("shardjob: stale shard epoch")
	// ErrStaleOffset means a checkpoint moved backwards inside one term.
	ErrStaleOffset = errors.New("shardjob: stale checkpoint offset")
	// ErrInjectedCrash marks the deterministic failure used by tests.
	ErrInjectedCrash = errors.New("shardjob: injected crash")
)

// Split assigns keys to shards by FNV-1a. The same key always lands on the same
// shard, and the result keeps input order inside each shard.
func Split(keys []string, shards int) ([][]string, error) {
	if shards <= 0 {
		return nil, fmt.Errorf("split %d keys: shards must be positive, got %d", len(keys), shards)
	}
	out := make([][]string, shards)
	for _, k := range keys {
		h := fnv.New32a()
		if _, err := h.Write([]byte(k)); err != nil {
			return nil, fmt.Errorf("hash %q: %w", k, err)
		}
		idx := int(h.Sum32() % uint32(shards))
		out[idx] = append(out[idx], k)
	}
	return out, nil
}

// Checkpoint stores the last committed offset per shard. Epoch is the shard
// lease term, so a worker that lost its lease cannot move the offset.
type Checkpoint struct {
	mu      sync.Mutex
	offsets map[string]int
	epochs  map[string]uint64
}

// NewCheckpoint builds an empty checkpoint store.
func NewCheckpoint() *Checkpoint {
	return &Checkpoint{offsets: make(map[string]int), epochs: make(map[string]uint64)}
}

// Load returns the committed offset for shard, zero when nothing is committed.
func (c *Checkpoint) Load(shard string) int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.offsets[shard]
}

// Commit advances shard to offset under epoch. Equal offset is idempotent;
// a lower epoch or a lower offset inside the same epoch is rejected.
func (c *Checkpoint) Commit(shard string, epoch uint64, offset int) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	cur := c.epochs[shard]
	if epoch < cur {
		return fmt.Errorf("commit %s epoch %d: %w", shard, epoch, ErrStaleEpoch)
	}
	if epoch == cur && offset < c.offsets[shard] {
		return fmt.Errorf("commit %s offset %d < %d: %w", shard, offset, c.offsets[shard], ErrStaleOffset)
	}
	c.epochs[shard] = epoch
	c.offsets[shard] = offset
	return nil
}

// Sink is the side-effect boundary. It deduplicates by business key, so a
// replayed batch cannot charge or ship the same order twice.
type Sink struct {
	mu         sync.Mutex
	seen       map[string]bool
	applied    int
	suppressed int
}

// NewSink builds an empty idempotent sink.
func NewSink() *Sink {
	return &Sink{seen: make(map[string]bool)}
}

// Apply performs the effect once per key and reports whether it was applied.
func (s *Sink) Apply(key string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.seen[key] {
		s.suppressed++
		return false
	}
	s.seen[key] = true
	s.applied++
	return true
}

// Counts returns applied and suppressed effect counts.
func (s *Sink) Counts() (applied, suppressed int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.applied, s.suppressed
}

// Runner executes one shard from its committed offset.
type Runner struct {
	Sink        *Sink
	CP          *Checkpoint
	Epoch       uint64
	CommitEvery int
}

// Result reports one run, including replays after a restart.
type Result struct {
	Attempts   int
	Applied    int
	Suppressed int
	Checkpoint int
}

// Run processes items from the committed offset. crashAfter > 0 stops the run
// after that many attempts, before the next checkpoint, emulating a process
// death between the side effect and the offset commit.
func (r *Runner) Run(shard string, items []string, crashAfter int) (Result, error) {
	start := r.CP.Load(shard)
	res := Result{Checkpoint: start}
	for i := start; i < len(items); i++ {
		res.Attempts++
		if r.Sink.Apply(items[i]) {
			res.Applied++
		} else {
			res.Suppressed++
		}
		if crashAfter > 0 && res.Attempts == crashAfter {
			return res, fmt.Errorf("crash after %d attempts at offset %d: %w", res.Attempts, i, ErrInjectedCrash)
		}
		if r.CommitEvery > 0 && (i+1-start)%r.CommitEvery == 0 {
			if err := r.CP.Commit(shard, r.Epoch, i+1); err != nil {
				return res, err
			}
			res.Checkpoint = i + 1
		}
	}
	if err := r.CP.Commit(shard, r.Epoch, len(items)); err != nil {
		return res, err
	}
	res.Checkpoint = len(items)
	return res, nil
}

=== RUN   TestRunCrashResume
    shard_test.go:40: shard=0 keys=250
    shard_test.go:40: shard=1 keys=250
    shard_test.go:40: shard=2 keys=250
    shard_test.go:40: shard=3 keys=250
    shard_test.go:126: shard0=250 crash_attempts=37 checkpoint=30 replay=7 applied=1000 suppressed=7
--- PASS: TestRunCrashResume (0.00s)