《后端架构设计·分布式与数据一致性》6.2 会话保证：读己之写、单调读与一致前缀

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

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

// Sentinel errors returned by the session layer.
var (
	ErrStaleReplica   = errors.New("session: replica is behind the session watermark")
	ErrWriteGap       = errors.New("session: write sequence gap")
	ErrDuplicateWrite = errors.New("session: duplicate write sequence")
)

// Entry is one committed write in the global log.
type Entry struct {
	Version uint64
	Key     string
	Value   string
}

// Replica stores a prefix of the log. Applied is the highest version applied,
// so a read at Applied=n sees exactly the first n writes.
type Replica struct {
	mu      sync.Mutex
	name    string
	applied uint64
	data    map[string]string
}

func newReplica(name string) *Replica {
	return &Replica{name: name, data: make(map[string]string)}
}

// Name returns the replica identifier.
func (r *Replica) Name() string { return r.name }

// Applied returns the highest log version applied on this replica.
func (r *Replica) Applied() uint64 {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.applied
}

func (r *Replica) apply(entries []Entry) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for _, e := range entries {
		r.data[e.Key] = e.Value
		r.applied = e.Version
	}
}

// Session carries the watermarks that turn per-replica reads into session
// guarantees. WriteWatermark is the version of the session's last write;
// ReadWatermark is the highest version any read in the session has observed.
type Session struct {
	mu             sync.Mutex
	Client         string
	lastWriteSeq   uint64
	writeWatermark uint64
	readWatermark  uint64
}

// NewSession creates an empty session for one client.
func NewSession(client string) *Session { return &Session{Client: client} }

// Watermarks returns the write and read watermarks as one atomic snapshot.
func (s *Session) Watermarks() (write, read uint64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.writeWatermark, s.readWatermark
}

// Cluster is a single-leader log with lagging followers. The replica map is
// immutable after construction, so readers may use it without the log lock.
type Cluster struct {
	mu       sync.Mutex
	log      []Entry
	replicas map[string]*Replica
	leader   string
	lastSeq  map[string]uint64
}

// NewCluster creates a leader plus the named followers.
func NewCluster(leader string, followers ...string) *Cluster {
	c := &Cluster{
		replicas: make(map[string]*Replica, len(followers)+1),
		leader:   leader,
		lastSeq:  make(map[string]uint64),
	}
	c.replicas[leader] = newReplica(leader)
	for _, name := range followers {
		c.replicas[name] = newReplica(name)
	}
	return c
}

// Leader returns the leader replica name.
func (c *Cluster) Leader() string { return c.leader }

// Write appends one entry with the session's next sequence number.
func (c *Cluster) Write(s *Session, key, value string) (uint64, error) {
	s.mu.Lock()
	seq := s.lastWriteSeq + 1
	s.mu.Unlock()
	return c.WriteWithSeq(s, seq, key, value)
}

// WriteWithSeq appends one entry with an explicit session sequence. The
// coordinator rejects duplicates and gaps, which enforces monotonic writes
// even when a client retries after a timeout.
func (c *Cluster) WriteWithSeq(s *Session, seq uint64, key, value string) (uint64, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	s.mu.Lock()
	defer s.mu.Unlock()

	if seq != s.lastWriteSeq+1 {
		if seq <= s.lastWriteSeq {
			return 0, fmt.Errorf("%w: client %s seq %d, last %d", ErrDuplicateWrite, s.Client, seq, s.lastWriteSeq)
		}
		return 0, fmt.Errorf("%w: client %s seq %d, last %d", ErrWriteGap, s.Client, seq, s.lastWriteSeq)
	}
	if last := c.lastSeq[s.Client]; seq <= last {
		return 0, fmt.Errorf("%w: coordinator has seq %d for %s", ErrDuplicateWrite, last, s.Client)
	}

	version := uint64(len(c.log)) + 1
	c.log = append(c.log, Entry{Version: version, Key: key, Value: value})
	c.lastSeq[s.Client] = seq
	s.lastWriteSeq = seq
	s.writeWatermark = version

	leader := c.replicas[c.leader]
	leader.apply(c.log[len(c.log)-1:])
	return version, nil
}

// Replicate applies every entry up to version upto on the named follower.
// Entries are applied in order, so the replica always holds a log prefix.
func (c *Cluster) Replicate(name string, upto uint64) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	r, ok := c.replicas[name]
	if !ok {
		return fmt.Errorf("session: unknown replica %q", name)
	}
	if upto > uint64(len(c.log)) {
		return fmt.Errorf("session: version %d beyond log length %d", upto, len(c.log))
	}
	r.mu.Lock()
	from := r.applied
	r.mu.Unlock()
	if upto <= from {
		return nil
	}
	r.apply(c.log[from:upto])
	return nil
}

// Read serves one key from a replica, refusing to move backwards in time.
// It enforces read-your-writes and monotonic reads simultaneously.
func (c *Cluster) Read(s *Session, replica, key string) (string, uint64, bool, error) {
	r, ok := c.replicas[replica]
	if !ok {
		return "", 0, false, fmt.Errorf("session: unknown replica %q", replica)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	need := s.writeWatermark
	if s.readWatermark > need {
		need = s.readWatermark
	}

	r.mu.Lock()
	defer r.mu.Unlock()
	if r.applied < need {
		return "", r.applied, false, fmt.Errorf("%w: need %d, replica %s has %d", ErrStaleReplica, need, replica, r.applied)
	}
	value, found := r.data[key]
	if r.applied > s.readWatermark {
		s.readWatermark = r.applied
	}
	return value, r.applied, found, nil
}

// Pick returns the first candidate that satisfies the session watermark, the
// routing half of sticky sessions.
func (c *Cluster) Pick(s *Session, candidates []string) (string, error) {
	write, read := s.Watermarks()
	need := write
	if read > need {
		need = read
	}
	for _, name := range candidates {
		r, ok := c.replicas[name]
		if !ok {
			continue
		}
		if r.Applied() >= need {
			return name, nil
		}
	}
	return "", fmt.Errorf("%w: no candidate has version %d", ErrStaleReplica, need)
}

package session

import (
	"errors"
	"testing"
)

func TestReadYourWrites(t *testing.T) {
	c := NewCluster("leader", "follower")
	s := NewSession("u1")

	version, err := c.Write(s, "order-status", "PAID")
	if err != nil {
		t.Fatalf("Write: %v", err)
	}
	if _, _, _, err := c.Read(s, "follower", "order-status"); !errors.Is(err, ErrStaleReplica) {
		t.Fatalf("lagging read err = %v, want ErrStaleReplica", err)
	}
	if err := c.Replicate("follower", version); err != nil {
		t.Fatalf("Replicate: %v", err)
	}
	got, ver, ok, err := c.Read(s, "follower", "order-status")
	if err != nil || !ok || got != "PAID" || ver != version {
		t.Fatalf("read = %q, %d, %v, %v; want PAID, %d, true, nil", got, ver, ok, err, version)
	}
}

func TestMonotonicReads(t *testing.T) {
	c := NewCluster("leader", "r1", "r2")
	writer := NewSession("u2")
	for _, kv := range [][2]string{{"a", "1"}, {"b", "2"}, {"c", "3"}} {
		if _, err := c.Write(writer, kv[0], kv[1]); err != nil {
			t.Fatalf("Write: %v", err)
		}
	}
	if err := c.Replicate("r1", 3); err != nil {
		t.Fatalf("Replicate r1: %v", err)
	}
	if err := c.Replicate("r2", 1); err != nil {
		t.Fatalf("Replicate r2: %v", err)
	}

	reader := NewSession("reader")
	if _, ver, _, err := c.Read(reader, "r1", "a"); err != nil || ver != 3 {
		t.Fatalf("first read = version %d, %v; want 3, nil", ver, err)
	}
	if _, _, _, err := c.Read(reader, "r2", "a"); !errors.Is(err, ErrStaleReplica) {
		t.Fatalf("second read err = %v, want ErrStaleReplica", err)
	}
}

$ go test ./article/session/ -count=1 -v
=== RUN   TestReadYourWrites
--- PASS: TestReadYourWrites (0.00s)
=== RUN   TestMonotonicReads
--- PASS: TestMonotonicReads (0.00s)
=== RUN   TestMonotonicWrites
--- PASS: TestMonotonicWrites (0.00s)
=== RUN   TestConsistentPrefix
--- PASS: TestConsistentPrefix (0.00s)
=== RUN   TestStickyRouteByWatermark
--- PASS: TestStickyRouteByWatermark (0.00s)
PASS
ok  	archlab/article/session	0.425s