《后端架构设计·分布式与数据一致性》3.1 单主复制：复制日志、位点与读己之写

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

import (
	"errors"
	"fmt"
)

var (
	ErrStaleReplica   = errors.New("replica has not applied the required position")
	ErrNoQuorum       = errors.New("write did not reach quorum")
	ErrUnknownSession = errors.New("unknown session")
)

type Entry struct {
	Index   uint64
	Command string
}

type SessionToken struct {
	SessionID string
	LSN       uint64
}

type Replica struct {
	ID           string
	Log          []Entry
	CommitIndex  uint64
	AppliedIndex uint64
}

func (r *Replica) LastIndex() uint64 {
	if len(r.Log) == 0 {
		return 0
	}
	return r.Log[len(r.Log)-1].Index
}

type Cluster struct {
	replicas   []*Replica
	primaryIdx int
	down       map[string]bool
	sessions   map[string]uint64
}

func NewCluster(ids ...string) (*Cluster, error) {
	if len(ids) < 3 {
		return nil, fmt.Errorf("cluster needs 3 replicas: got %d", len(ids))
	}
	c := &Cluster{down: map[string]bool{}, sessions: map[string]uint64{}}
	for _, id := range ids {
		c.replicas = append(c.replicas, &Replica{ID: id})
	}
	return c, nil
}

func (c *Cluster) quorum() int { return len(c.replicas)/2 + 1 }

func (c *Cluster) Primary() *Replica { return c.replicas[c.primaryIdx] }

func (c *Cluster) SetDown(id string, down bool) { c.down[id] = down }

// Write returns a token only after a majority stores the entry.
func (c *Cluster) Write(sessionID, command string) (SessionToken, error) {
	leader := c.Primary()
	if c.down[leader.ID] {
		return SessionToken{}, fmt.Errorf("leader %s is down", leader.ID)
	}
	entry := Entry{Index: leader.LastIndex() + 1, Command: command}
	leader.Log = append(leader.Log, entry)
	acks := 1
	for _, r := range c.replicas {
		if r == leader || c.down[r.ID] {
			continue
		}
		r.Log = append(r.Log, entry)
		acks++
	}
	if acks < c.quorum() {
		return SessionToken{}, fmt.Errorf("%w: acks=%d quorum=%d", ErrNoQuorum, acks, c.quorum())
	}
	leader.CommitIndex = entry.Index
	for _, r := range c.replicas {
		if r != leader && r.LastIndex() >= entry.Index {
			r.CommitIndex = entry.Index
		}
	}
	c.sessions[sessionID] = entry.Index
	return SessionToken{SessionID: sessionID, LSN: entry.Index}, nil
}

func (c *Cluster) Apply(replicaID string) (uint64, error) {
	for _, r := range c.replicas {
		if r.ID == replicaID {
			r.AppliedIndex = r.CommitIndex
			return r.AppliedIndex, nil
		}
	}
	return 0, fmt.Errorf("unknown replica %q", replicaID)
}

// Read enforces the session's minimum visible position.
func (c *Cluster) Read(sessionID, replicaID string) (string, error) {
	lsn, ok := c.sessions[sessionID]
	if !ok {
		return "", fmt.Errorf("%w: %s", ErrUnknownSession, sessionID)
	}
	for _, r := range c.replicas {
		if r.ID != replicaID {
			continue
		}
		if r.AppliedIndex < lsn {
			return "", fmt.Errorf("%w: %s applied=%d required=%d", ErrStaleReplica, r.ID, r.AppliedIndex, lsn)
		}
		for _, e := range r.Log {
			if e.Index == lsn {
				return e.Command, nil
			}
		}
		return "", fmt.Errorf("%w: lsn=%d missing", ErrStaleReplica, lsn)
	}
	return "", fmt.Errorf("unknown replica %q", replicaID)
}

// Failover promotes the reachable replica with the longest log.
func (c *Cluster) Failover() (int, error) {
	up, best := 0, -1
	for i, r := range c.replicas {
		if c.down[r.ID] {
			continue
		}
		up++
		if best == -1 || r.LastIndex() > c.replicas[best].LastIndex() ||
			(r.LastIndex() == c.replicas[best].LastIndex() && r.ID < c.replicas[best].ID) {
			best = i
		}
	}
	if up < c.quorum() {
		return 0, fmt.Errorf("%w: up=%d quorum=%d", ErrNoQuorum, up, c.quorum())
	}
	leader := c.replicas[best]
	c.primaryIdx = best
	dropped := 0
	for _, r := range c.replicas {
		if r.ID == leader.ID || c.down[r.ID] || len(r.Log) <= len(leader.Log) {
			continue
		}
		dropped += len(r.Log) - len(leader.Log)
		r.Log = r.Log[:len(leader.Log)]
		if r.CommitIndex > leader.LastIndex() {
			r.CommitIndex = leader.LastIndex()
		}
		if r.AppliedIndex > r.CommitIndex {
			r.AppliedIndex = r.CommitIndex
		}
	}
	commit := uint64(0)
	for idx := leader.LastIndex(); idx > 0; idx-- {
		count := 0
		for _, r := range c.replicas {
			if r.LastIndex() >= idx {
				count++
			}
		}
		if count >= c.quorum() {
			commit = idx
			break
		}
	}
	leader.CommitIndex = commit
	if leader.AppliedIndex > commit {
		leader.AppliedIndex = commit
	}
	return dropped, nil
}

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

import (
	"errors"
	"testing"
)

func TestSessionRead(t *testing.T) {
	tests := []struct {
		name    string
		apply   bool
		wantErr error
		want    string
	}{
		{"lagging follower rejects", false, ErrStaleReplica, ""},
		{"applied follower serves", true, nil, "order-1001:CREATED"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c, err := NewCluster("n1", "n2", "n3")
			if err != nil {
				t.Fatalf("NewCluster: %v", err)
			}
			if _, err := c.Write("s1", "order-1001:CREATED"); err != nil {
				t.Fatalf("Write: %v", err)
			}
			if tt.apply {
				if _, err := c.Apply("n2"); err != nil {
					t.Fatalf("Apply: %v", err)
				}
			}
			got, err := c.Read("s1", "n2")
			if tt.wantErr != nil {
				if !errors.Is(err, tt.wantErr) {
					t.Fatalf("Read error = %v, want %v", err, tt.wantErr)
				}
				return
			}
			if err != nil || got != tt.want {
				t.Fatalf("Read = %q, %v; want %q, nil", got, err, tt.want)
			}
		})
	}
}

func TestQuorumAndFailoverBoundary(t *testing.T) {
	c, _ := NewCluster("n1", "n2", "n3")
	session := "s1"
	if _, err := c.Write(session, "order-1001:CREATED"); err != nil {
		t.Fatalf("first write: %v", err)
	}

	// A write that reaches only the leader is not acknowledged.
	c.SetDown("n2", true)
	c.SetDown("n3", true)
	if _, err := c.Write(session, "order-1001:PAID"); !errors.Is(err, ErrNoQuorum) {
		t.Fatalf("second write error = %v, want ErrNoQuorum", err)
	}

	// A majority elects n2, which never saw the uncommitted tail.
	c.SetDown("n1", true)
	c.SetDown("n2", false)
	c.SetDown("n3", false)
	if _, err := c.Failover(); err != nil {
		t.Fatalf("Failover: %v", err)
	}
	if got := c.Primary().CommitIndex; got != 1 {
		t.Fatalf("commit index = %d, want 1", got)
	}
	if _, err := c.Apply("n2"); err != nil {
		t.Fatalf("Apply: %v", err)
	}
	got, err := c.Read(session, "n2")
	if err != nil || got != "order-1001:CREATED" {
		t.Fatalf("Read = %q, %v; want CREATED, nil", got, err)
	}
}

$ go test ./article/replication/ -count=1
--- PASS: TestSessionRead (0.00s)
    --- PASS: TestSessionRead/lagging_follower_rejects (0.00s)
    --- PASS: TestSessionRead/applied_follower_serves (0.00s)
--- PASS: TestQuorumAndFailoverBoundary (0.00s)
PASS
ok  	archlab/article/replication	0.241s