《后端架构设计·分布式与数据一致性》3.2 多主与无主复制：冲突检测与合并

// file: internal/platform/multileader/multileader.go
// tier: L1
// verify: go test ./internal/platform/multileader/ -count=1
package multileader

import "fmt"

type Order int

const (
	Equal Order = iota
	Before
	After
	Concurrent
)

type VersionVector map[string]uint64

func (v VersionVector) Increment(node string) { v[node]++ }

// Merge returns the component-wise maximum of two vectors.
func (v VersionVector) Merge(other VersionVector) VersionVector {
	out := make(VersionVector, len(v)+len(other))
	for k, val := range v {
		out[k] = val
	}
	for k, val := range other {
		if val > out[k] {
			out[k] = val
		}
	}
	return out
}

// Compare tells concurrent updates apart from causally ordered ones.
func (v VersionVector) Compare(other VersionVector) Order {
	less, greater := false, false
	for k, val := range v {
		if oval := other[k]; val < oval {
			less = true
		} else if val > oval {
			greater = true
		}
	}
	for k, oval := range other {
		if _, ok := v[k]; !ok && oval > 0 {
			less = true
		}
	}
	switch {
	case less && greater:
		return Concurrent
	case less:
		return Before
	case greater:
		return After
	default:
		return Equal
	}
}

type LWWRegister struct {
	Value     string
	Timestamp int64
	Node      string
}

// MergeLWW keeps the greater timestamp; equal timestamps use the node ID.
func MergeLWW(a, b LWWRegister) LWWRegister {
	if a.Timestamp != b.Timestamp {
		if a.Timestamp > b.Timestamp {
			return a
		}
		return b
	}
	if a.Node >= b.Node {
		return a
	}
	return b
}

// PNCounter keeps per-node increments and decrements. Merge takes the maximum.
type PNCounter struct {
	inc map[string]int64
	dec map[string]int64
}

func NewPNCounter() *PNCounter {
	return &PNCounter{inc: map[string]int64{}, dec: map[string]int64{}}
}

func (c *PNCounter) Add(node string, delta int64) {
	if delta >= 0 {
		c.inc[node] += delta
		return
	}
	c.dec[node] -= delta
}

func (c *PNCounter) Merge(other *PNCounter) {
	for node, val := range other.inc {
		if val > c.inc[node] {
			c.inc[node] = val
		}
	}
	for node, val := range other.dec {
		if val > c.dec[node] {
			c.dec[node] = val
		}
	}
}

func (c *PNCounter) Value() int64 {
	var total int64
	for _, val := range c.inc {
		total += val
	}
	for _, val := range c.dec {
		total -= val
	}
	return total
}

// ORSet is add-wins: a remove only tombstones tags it has observed.
type ORSet struct {
	adds    map[string]map[string]struct{}
	removed map[string]map[string]struct{}
	seq     int
}

func NewORSet() *ORSet {
	return &ORSet{adds: map[string]map[string]struct{}{}, removed: map[string]map[string]struct{}{}}
}

func (s *ORSet) Add(node, value string) string {
	s.seq++
	tag := fmt.Sprintf("%s:%d", node, s.seq)
	if s.adds[value] == nil {
		s.adds[value] = map[string]struct{}{}
	}
	s.adds[value][tag] = struct{}{}
	return tag
}

func (s *ORSet) Remove(value string) {
	if s.removed[value] == nil {
		s.removed[value] = map[string]struct{}{}
	}
	for tag := range s.adds[value] {
		s.removed[value][tag] = struct{}{}
	}
}

func (s *ORSet) Contains(value string) bool {
	for tag := range s.adds[value] {
		if _, gone := s.removed[value][tag]; !gone {
			return true
		}
	}
	return false
}

func (s *ORSet) Merge(other *ORSet) {
	for value, tags := range other.adds {
		if s.adds[value] == nil {
			s.adds[value] = map[string]struct{}{}
		}
		for tag := range tags {
			s.adds[value][tag] = struct{}{}
		}
	}
	for value, tags := range other.removed {
		if s.removed[value] == nil {
			s.removed[value] = map[string]struct{}{}
		}
		for tag := range tags {
			s.removed[value][tag] = struct{}{}
		}
	}
	if other.seq > s.seq {
		s.seq = other.seq
	}
}

// QuorumsOverlap is true exactly when R+W>N.
func QuorumsOverlap(n, r, w int) bool {
	if n < 1 || r < 1 || w < 1 || r > n || w > n {
		return false
	}
	return r+w > n
}

// file: internal/platform/multileader/multileader_test.go
// tier: L1
// verify: go test ./internal/platform/multileader/ -count=1
package multileader

import "testing"

func TestConflictDetection(t *testing.T) {
	tests := []struct {
		name string
		a, b VersionVector
		want Order
	}{
		{"equal", VersionVector{"a": 1}, VersionVector{"a": 1}, Equal},
		{"before", VersionVector{"a": 1}, VersionVector{"a": 2}, Before},
		{"concurrent", VersionVector{"a": 2, "b": 1}, VersionVector{"a": 1, "b": 2}, Concurrent},
		{"missing key", VersionVector{"a": 1}, VersionVector{"a": 1, "b": 1}, Before},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := tt.a.Compare(tt.b); got != tt.want {
				t.Fatalf("Compare = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestLWWPicksTimestamp(t *testing.T) {
	newer := LWWRegister{Value: "上海市浦东新区世纪大道 100 号", Timestamp: 1000, Node: "b"}
	older := LWWRegister{Value: "上海市徐汇区漕溪北路 88 号", Timestamp: 900, Node: "a"}
	if got := MergeLWW(older, newer); got != newer {
		t.Fatalf("MergeLWW = %+v, want %+v", got, newer)
	}
	if MergeLWW(newer, older) != MergeLWW(older, newer) {
		t.Fatalf("MergeLWW is not commutative")
	}
}

func TestCRDTMerge(t *testing.T) {
	a, b := NewPNCounter(), NewPNCounter()
	a.Add("n1", 3)
	b.Add("n2", 4)
	b.Add("n2", -1)
	a.Merge(b)
	if got := a.Value(); got != 6 {
		t.Fatalf("PNCounter = %d, want 6", got)
	}
	before := a.Value()
	a.Merge(b)
	if a.Value() != before {
		t.Fatalf("PNCounter merge is not idempotent")
	}

	x, y := NewORSet(), NewORSet()
	x.Add("n1", "gift-wrap")
	y.Add("n2", "gift-wrap")
	x.Remove("gift-wrap")
	x.Merge(y)
	if !x.Contains("gift-wrap") {
		t.Fatalf("concurrent add should survive a remove")
	}
}

func TestQuorumOverlap(t *testing.T) {
	for _, tt := range []struct {
		n, r, w int
		want    bool
	}{
		{3, 2, 2, true},
		{3, 1, 2, false},
		{5, 3, 3, true},
		{5, 2, 3, false},
	} {
		if got := QuorumsOverlap(tt.n, tt.r, tt.w); got != tt.want {
			t.Fatalf("QuorumsOverlap(%d,%d,%d) = %v, want %v", tt.n, tt.r, tt.w, got, tt.want)
		}
	}
}

$ go test ./article/multileader/ -count=1
--- PASS: TestConflictDetection (0.00s)
--- PASS: TestLWWPicksTimestamp (0.00s)
--- PASS: TestCRDTMerge (0.00s)
--- PASS: TestQuorumOverlap (0.00s)
PASS
ok  	archlab/article/multileader	0.261s