《后端架构设计·分布式与数据一致性》6.1 线性一致、顺序一致与因果一致

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

import "fmt"

// Kind distinguishes reads from writes in a recorded history.
type Kind int

// Operation kinds. A history is a sequence of reads and writes on typed keys.
const (
	Read Kind = iota
	Write
)

// Op is one completed operation. Invoke and Response are logical timestamps;
// two operations are concurrent exactly when their intervals overlap.
type Op struct {
	Client    string
	Key       string
	Kind      Kind
	Value     int
	Invoke    int64
	Response  int64
	ReadsFrom int // index of the observed write, -1 for the initial value
}

// History is a recorded execution, ordered by invocation time. Operations of
// one client are assumed to appear in program order and never overlap.
type History []Op

func (h History) validate() error {
	for i, op := range h {
		if op.Key == "" {
			return fmt.Errorf("consistency: op %d has empty key", i)
		}
		if op.Response < op.Invoke {
			return fmt.Errorf("consistency: op %d ends before it starts", i)
		}
		if op.Kind == Read && op.ReadsFrom >= len(h) {
			return fmt.Errorf("consistency: read %d points outside the history", i)
		}
	}
	return nil
}

// Linearizable reports whether some total order of all operations both
// respects real time (a response before a later invocation) and makes every
// read return the value of the latest write before it.
func (h History) Linearizable(initial map[string]int) (bool, error) {
	if err := h.validate(); err != nil {
		return false, err
	}
	n := len(h)
	precedes := make([][]bool, n)
	for i := range precedes {
		precedes[i] = make([]bool, n)
	}
	for i := range h {
		for j := range h {
			if i != j && h[i].Response < h[j].Invoke {
				precedes[i][j] = true
			}
		}
	}
	return h.search(initial, precedes), nil
}

// Sequential reports whether some total order respects each client's program
// order, without requiring that order to match real time.
func (h History) Sequential(initial map[string]int) (bool, error) {
	if err := h.validate(); err != nil {
		return false, err
	}
	n := len(h)
	precedes := make([][]bool, n)
	for i := range precedes {
		precedes[i] = make([]bool, n)
	}
	last := make(map[string]int, n)
	for i, op := range h {
		if prev, ok := last[op.Client]; ok {
			precedes[prev][i] = true
		}
		last[op.Client] = i
	}
	return h.search(initial, precedes), nil
}

// search backtracks over every operation whose predecessors are already
// placed, applying register semantics. It is exponential in the worst case
// and is intended for small recorded histories, not production traffic.
func (h History) search(initial map[string]int, precedes [][]bool) bool {
	state := make(map[string]int, len(initial))
	for k, v := range initial {
		state[k] = v
	}
	placed := make([]bool, len(h))

	var walk func(done int) bool
	walk = func(done int) bool {
		if done == len(h) {
			return true
		}
		for i := range h {
			if placed[i] {
				continue
			}
			blocked := false
			for j := range h {
				if !placed[j] && precedes[j][i] {
					blocked = true
					break
				}
			}
			if blocked {
				continue
			}

			prev, had := state[h[i].Key]
			ok := true
			if h[i].Kind == Write {
				state[h[i].Key] = h[i].Value
			} else {
				ok = h[i].Value == prev
			}
			placed[i] = true
			if ok && walk(done+1) {
				return true
			}
			placed[i] = false
			if had {
				state[h[i].Key] = prev
			} else {
				delete(state, h[i].Key)
			}
		}
		return false
	}
	return walk(0)
}

// CausallyConsistent reports whether the annotated history admits a causal
// order: session order plus reads-from edges must be acyclic, and a read may
// not skip a write that is causally after the write it observed.
func (h History) CausallyConsistent() (bool, error) {
	if err := h.validate(); err != nil {
		return false, err
	}
	n := len(h)
	adj := make([][]bool, n)
	for i := range adj {
		adj[i] = make([]bool, n)
	}
	last := make(map[string]int, n)
	for i, op := range h {
		if prev, ok := last[op.Client]; ok {
			adj[prev][i] = true
		}
		last[op.Client] = i
	}
	for i, op := range h {
		if op.Kind != Read || op.ReadsFrom < 0 {
			continue
		}
		w := h[op.ReadsFrom]
		if w.Kind != Write || w.Key != op.Key {
			return false, fmt.Errorf("consistency: read %d observes non-write or other key", i)
		}
		adj[op.ReadsFrom][i] = true
	}

	for k := 0; k < n; k++ {
		for i := 0; i < n; i++ {
			if !adj[i][k] {
				continue
			}
			for j := 0; j < n; j++ {
				if adj[k][j] {
					adj[i][j] = true
				}
			}
		}
	}
	for i := 0; i < n; i++ {
		if adj[i][i] {
			return false, nil
		}
	}

	for i, op := range h {
		if op.Kind != Read || op.ReadsFrom < 0 {
			continue
		}
		for w := range h {
			if w == op.ReadsFrom || h[w].Kind != Write || h[w].Key != op.Key {
				continue
			}
			if adj[op.ReadsFrom][w] && adj[w][i] {
				return false, nil
			}
		}
	}
	return true, nil
}

package consistency

import "testing"

func TestModelBoundaries(t *testing.T) {
	linear := History{
		{Client: "a", Key: "stock", Kind: Write, Value: 1, Invoke: 0, Response: 1},
		{Client: "b", Key: "stock", Kind: Read, Value: 1, Invoke: 2, Response: 3},
	}
	sequentialOnly := History{
		{Client: "p1", Key: "x", Kind: Write, Value: 1, Invoke: 0, Response: 1},
		{Client: "p2", Key: "x", Kind: Write, Value: 2, Invoke: 0, Response: 1},
		{Client: "p3", Key: "x", Kind: Read, Value: 1, Invoke: 2, Response: 3},
		{Client: "p4", Key: "x", Kind: Read, Value: 2, Invoke: 2, Response: 3},
	}
	staleCausalRead := History{
		{Client: "a", Key: "s", Kind: Write, Value: 1, Invoke: 0, Response: 1, ReadsFrom: -1},
		{Client: "a", Key: "s", Kind: Write, Value: 2, Invoke: 2, Response: 3, ReadsFrom: -1},
		{Client: "c", Key: "s", Kind: Read, Value: 2, Invoke: 4, Response: 5, ReadsFrom: 1},
		{Client: "c", Key: "s", Kind: Read, Value: 1, Invoke: 6, Response: 7, ReadsFrom: 0},
	}
	if ok, err := linear.Linearizable(nil); err != nil || !ok {
		t.Fatalf("linear history = %v, %v; want true, nil", ok, err)
	}
	if ok, err := sequentialOnly.Linearizable(nil); err != nil || ok {
		t.Fatalf("sequential history is linearizable = %v, %v; want false, nil", ok, err)
	}
	if ok, err := sequentialOnly.Sequential(nil); err != nil || !ok {
		t.Fatalf("sequential history = %v, %v; want true, nil", ok, err)
	}
	if ok, err := staleCausalRead.CausallyConsistent(); err != nil || ok {
		t.Fatalf("stale causal read = %v, %v; want false, nil", ok, err)
	}
}

$ go test ./article/consistency/ -count=1 -v
=== RUN   TestLinearizableHistory
=== RUN   TestLinearizableHistory/read_sees_a_completed_write
=== RUN   TestLinearizableHistory/stale_read_after_the_write_completed
=== RUN   TestLinearizableHistory/concurrent_read_may_observe_the_old_value
--- PASS: TestLinearizableHistory (0.00s)
    --- PASS: TestLinearizableHistory/read_sees_a_completed_write (0.00s)
    --- PASS: TestLinearizableHistory/stale_read_after_the_write_completed (0.00s)
    --- PASS: TestLinearizableHistory/concurrent_read_may_observe_the_old_value (0.00s)
=== RUN   TestSequentialWithoutRealTime
--- PASS: TestSequentialWithoutRealTime (0.00s)
=== RUN   TestCausalConcurrentWrites
--- PASS: TestCausalConcurrentWrites (0.00s)
=== RUN   TestCausalRejectsSkippedWrite
--- PASS: TestCausalRejectsSkippedWrite (0.00s)
=== RUN   TestModelBoundaries
--- PASS: TestModelBoundaries (0.00s)
PASS
ok  	archlab/article/consistency	0.263s