《后端架构设计·分布式与数据一致性》12.2 数据同步与冲突：双向复制、回环与归属

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

import (
	"errors"
	"fmt"
)

// ErrNotOwner means an event was produced by a site that does not own the
// order. Such an event must never overwrite the owner's data.
var ErrNotOwner = errors.New("replication: writer is not the home site")

// SiteID identifies one data center.
type SiteID string

// Vector counts observed mutations per origin site.
type Vector map[SiteID]uint64

// Clone returns an independent copy.
func (v Vector) Clone() Vector {
	out := make(Vector, len(v))
	for site, n := range v {
		out[site] = n
	}
	return out
}

// Get returns the counter for site. A missing site reads as zero.
func (v Vector) Get(site SiteID) uint64 {
	return v[site]
}

// Dominates reports whether v has seen every mutation recorded in other.
func (v Vector) Dominates(other Vector) bool {
	for site, n := range other {
		if v[site] < n {
			return false
		}
	}
	return true
}

// Equal reports component-wise equality.
func (v Vector) Equal(other Vector) bool {
	return v.Dominates(other) && other.Dominates(v)
}

// Concurrent reports whether neither vector dominates the other.
func Concurrent(a, b Vector) bool {
	return !a.Dominates(b) && !b.Dominates(a)
}

// Merge returns the component-wise maximum.
func Merge(a, b Vector) Vector {
	out := a.Clone()
	for site, n := range b {
		if n > out[site] {
			out[site] = n
		}
	}
	return out
}

// Event is one replication message. Path records every site that already
// applied it so a relayed event cannot loop forever.
type Event struct {
	ID      string
	OrderID string
	Origin  SiteID
	Path    []SiteID
	Version Vector
	Payload string
}

// Kind classifies the outcome of Receive.
type Kind string

const (
	KindApplied   Kind = "applied"
	KindDuplicate Kind = "duplicate"
	KindLoop      Kind = "loop"
	KindStale     Kind = "stale"
	KindConflict  Kind = "conflict"
	KindNotOwner  Kind = "not_owner"
)

// Record is the stored state of one order.
type Record struct {
	Payload string
	Version Vector
}

// Replica is one site's copy of the order data plus its ownership view.
type Replica struct {
	Site SiteID
	home map[string]SiteID
	data map[string]Record
	seen map[string]struct{}

	Applied   int
	Duplicate int
	Loop      int
	Stale     int
	Conflict  int
	NotOwner  int
}

// New returns an empty replica. home must map every known order to one site.
func New(site SiteID, home map[string]SiteID) (*Replica, error) {
	if site == "" {
		return nil, fmt.Errorf("replication: empty site id")
	}
	for order, owner := range home {
		if order == "" || owner == "" {
			return nil, fmt.Errorf("replication: invalid ownership entry %q -> %q", order, owner)
		}
	}
	r := &Replica{
		Site: site,
		home: make(map[string]SiteID, len(home)),
		data: make(map[string]Record),
		seen: make(map[string]struct{}),
	}
	for order, owner := range home {
		r.home[order] = owner
	}
	return r, nil
}

// LocalWrite accepts a write only at the order's home site and returns the
// event to replicate. The local copy is updated before the caller sends it.
func (r *Replica) LocalWrite(orderID, payload string) (Event, error) {
	if r.home[orderID] != r.Site {
		return Event{}, fmt.Errorf("replication: site %s writing order %s owned by %s: %w",
			r.Site, orderID, r.home[orderID], ErrNotOwner)
	}
	cur := r.data[orderID]
	version := cur.Version.Clone()
	if version == nil {
		version = Vector{}
	}
	version[r.Site]++
	id := fmt.Sprintf("%s-%s-%d", orderID, r.Site, version[r.Site])
	event := Event{
		ID:      id,
		OrderID: orderID,
		Origin:  r.Site,
		Path:    []SiteID{r.Site},
		Version: version.Clone(),
		Payload: payload,
	}
	r.data[orderID] = Record{Payload: payload, Version: version}
	r.seen[id] = struct{}{}
	r.Applied++
	return event, nil
}

// Receive applies an event from a peer. It never applies data from a
// non-owner, a stale version, or a path that already visited this site. A
// successfully applied event yields a relay event with this site appended to
// its path and a new message id, mirroring per-hop message ids in real relays.
func (r *Replica) Receive(event Event) (Kind, *Event, error) {
	if event.ID == "" || event.OrderID == "" || event.Origin == "" {
		return "", nil, fmt.Errorf("replication: incomplete event %+v", event)
	}
	for _, site := range event.Path {
		if site == r.Site {
			r.Loop++
			return KindLoop, nil, nil
		}
	}
	if _, ok := r.seen[event.ID]; ok {
		r.Duplicate++
		return KindDuplicate, nil, nil
	}
	if r.home[event.OrderID] != event.Origin {
		r.NotOwner++
		return KindNotOwner, nil, nil
	}
	cur, exists := r.data[event.OrderID]
	if exists {
		switch {
		case event.Version.Equal(cur.Version):
			if event.Payload == cur.Payload {
				r.Duplicate++
				return KindDuplicate, nil, nil
			}
			r.Conflict++
			return KindConflict, nil, nil
		case cur.Version.Dominates(event.Version):
			r.Stale++
			return KindStale, nil, nil
		case !event.Version.Dominates(cur.Version):
			r.Conflict++
			return KindConflict, nil, nil
		}
	}
	r.data[event.OrderID] = Record{Payload: event.Payload, Version: event.Version.Clone()}
	r.seen[event.ID] = struct{}{}
	r.Applied++

	path := append(append([]SiteID(nil), event.Path...), r.Site)
	relay := Event{
		ID:      event.ID + "@" + string(r.Site),
		OrderID: event.OrderID,
		Origin:  event.Origin,
		Path:    path,
		Version: event.Version.Clone(),
		Payload: event.Payload,
	}
	return KindApplied, &relay, nil
}

// Lookup returns the stored record for an order.
func (r *Replica) Lookup(orderID string) (Record, bool) {
	record, ok := r.data[orderID]
	return record, ok
}

$ go test ./internal/platform/replication/ -v -count=1
=== RUN   TestBidirectionalLoopSuppressed
    bidirectional_test.go:41: loop_suppressed a_applied=1 b_applied=1 a_loop=1 payload=PAID
=== RUN   TestDuplicateDeliveryIsIdempotent
    bidirectional_test.go:64: duplicate_delivery b_applied=1 b_duplicate=1
=== RUN   TestNonOwnerWriteRejected
    bidirectional_test.go:89: non_owner local_write_rejected=true remote_event=not_owner
=== RUN   TestStaleEventIgnored
    bidirectional_test.go:119: stale_ignored payload=FULFILLED stale_count=1
=== RUN   TestConcurrentVersionsDetected
    bidirectional_test.go:163: concurrent_detected merged=map[site-a:2 site-b:1] conflict_kept=FULFILLED
=== RUN   TestVectorDominanceTable
    bidirectional_test.go:193: vectors a=map[s1:2] b=map[s1:1] relation=a>b
    bidirectional_test.go:193: vectors a=map[s1:2] b=map[s1:2] relation=equal
    bidirectional_test.go:193: vectors a=map[s1:2] b=map[s2:1] relation=concurrent
    bidirectional_test.go:193: vectors a=map[s1:1] b=map[s2:1] relation=concurrent
PASS
ok  	archlab/internal/platform/replication	0.258s