《后端架构设计·分布式与数据一致性》1.2 故障模型：崩溃、丢包、分区与拜占庭

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

import (
	"errors"
	"fmt"
)

// Errors returned by the simulated network and replicas.
var (
	ErrCrashed     = errors.New("faultnet: node crashed")
	ErrUnknownNode = errors.New("faultnet: unknown node")
)

// Message is one in-flight request or reply.
type Message struct {
	From    int
	To      int
	Payload string
}

// Stats records what happened to messages, never what they meant.
type Stats struct {
	Sent      int
	Delivered int
	Dropped   int
	Corrupted int
}

// Network delivers messages in FIFO order with injectable faults.
type Network struct {
	nodes    int
	crashed  map[int]bool
	cut      map[[2]int]bool
	queue    []Message
	dropN    int
	corruptN int
	Stats    Stats
}

// New creates a fully connected network of n nodes.
func New(n int) *Network {
	return &Network{
		nodes:   n,
		crashed: make(map[int]bool),
		cut:     make(map[[2]int]bool),
	}
}

func (nw *Network) valid(node int) bool { return node >= 0 && node < nw.nodes }

// Send enqueues a message. A crashed sender cannot send.
func (nw *Network) Send(from, to int, payload string) error {
	if !nw.valid(from) || !nw.valid(to) {
		return fmt.Errorf("send %d->%d: %w", from, to, ErrUnknownNode)
	}
	if nw.crashed[from] {
		return fmt.Errorf("send %d->%d: %w", from, to, ErrCrashed)
	}
	nw.Stats.Sent++
	nw.queue = append(nw.queue, Message{From: from, To: to, Payload: payload})
	return nil
}

// DropNext loses the next n messages at delivery time.
func (nw *Network) DropNext(n int) { nw.dropN += n }

// CorruptNext rewrites the payload of the next n delivered messages.
func (nw *Network) CorruptNext(n int) { nw.corruptN += n }

// Partition cuts both directions between a and b.
func (nw *Network) Partition(a, b int) {
	nw.cut[[2]int{a, b}] = true
	nw.cut[[2]int{b, a}] = true
}

// Heal restores both directions between a and b.
func (nw *Network) Heal(a, b int) {
	delete(nw.cut, [2]int{a, b})
	delete(nw.cut, [2]int{b, a})
}

// Crash stops a node; its queued messages are dropped on delivery.
func (nw *Network) Crash(node int) { nw.crashed[node] = true }

// Recover restarts a node with an empty volatile state.
func (nw *Network) Recover(node int) { delete(nw.crashed, node) }

// Pending reports how many messages are still queued.
func (nw *Network) Pending() int { return len(nw.queue) }

// Step removes and returns the next deliverable message.
func (nw *Network) Step() (Message, bool) {
	for len(nw.queue) > 0 {
		m := nw.queue[0]
		nw.queue = nw.queue[1:]

		if nw.dropN > 0 {
			nw.dropN--
			nw.Stats.Dropped++
			continue
		}
		if nw.crashed[m.From] || nw.crashed[m.To] {
			nw.Stats.Dropped++
			continue
		}
		if nw.cut[[2]int{m.From, m.To}] {
			nw.Stats.Dropped++
			continue
		}
		if nw.corruptN > 0 {
			nw.corruptN--
			m.Payload = "corrupt:" + m.Payload
			nw.Stats.Corrupted++
		}
		nw.Stats.Delivered++
		return m, true
	}
	return Message{}, false
}

// Replica is one inventory state machine with a durable applied set.
type Replica struct {
	stable map[string]bool
	live   map[string]bool
	up     bool
}

// NewReplica starts an empty, running replica.
func NewReplica() *Replica {
	return &Replica{stable: make(map[string]bool), live: make(map[string]bool), up: true}
}

// Apply reserves exactly once per order ID and reports whether it was new.
func (r *Replica) Apply(orderID string) (bool, error) {
	if !r.up {
		return false, fmt.Errorf("apply %s: %w", orderID, ErrCrashed)
	}
	if r.stable[orderID] {
		r.live[orderID] = true
		return false, nil
	}
	r.stable[orderID] = true
	r.live[orderID] = true
	return true, nil
}

// Crash discards volatile state but keeps stable state.
func (r *Replica) Crash() {
	r.up = false
	r.live = make(map[string]bool)
}

// Recover reloads the working set from stable storage.
func (r *Replica) Recover() {
	r.up = true
	r.live = make(map[string]bool)
	for id := range r.stable {
		r.live[id] = true
	}
}

// Applied reports whether the order is present after the last recovery.
func (r *Replica) Applied(orderID string) bool { return r.live[orderID] }

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

import (
	"errors"
	"testing"
)

func TestFaultInjectionTable(t *testing.T) {
	cases := []struct {
		name        string
		inject      func(*Network)
		wantOK      bool
		wantDrop    int
		wantCorr    int
		wantPayload string
	}{
		{
			name:        "clean link",
			inject:      func(*Network) {},
			wantOK:      true,
			wantPayload: "reserve:ord-1",
		},
		{
			name:        "one message lost",
			inject:      func(nw *Network) { nw.DropNext(1) },
			wantOK:      false,
			wantDrop:    1,
			wantPayload: "",
		},
		{
			name:        "partition blocks delivery",
			inject:      func(nw *Network) { nw.Partition(0, 1) },
			wantOK:      false,
			wantDrop:    1,
			wantPayload: "",
		},
		{
			name:        "receiver crashed",
			inject:      func(nw *Network) { nw.Crash(1) },
			wantOK:      false,
			wantDrop:    1,
			wantPayload: "",
		},
		{
			name:        "byzantine payload rewrite",
			inject:      func(nw *Network) { nw.CorruptNext(1) },
			wantOK:      true,
			wantCorr:    1,
			wantPayload: "corrupt:reserve:ord-1",
		},
		{
			name:        "healed partition delivers",
			inject:      func(nw *Network) { nw.Partition(0, 1); nw.Heal(0, 1) },
			wantOK:      true,
			wantPayload: "reserve:ord-1",
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			nw := New(3)
			tc.inject(nw)
			if err := nw.Send(0, 1, "reserve:ord-1"); err != nil {
				t.Fatalf("send: %v", err)
			}
			msg, ok := nw.Step()
			if ok != tc.wantOK {
				t.Fatalf("delivered = %v, want %v", ok, tc.wantOK)
			}
			if tc.wantOK && msg.Payload != tc.wantPayload {
				t.Fatalf("payload = %q, want %q", msg.Payload, tc.wantPayload)
			}
			if nw.Stats.Dropped != tc.wantDrop {
				t.Fatalf("dropped = %d, want %d", nw.Stats.Dropped, tc.wantDrop)
			}
			if nw.Stats.Corrupted != tc.wantCorr {
				t.Fatalf("corrupted = %d, want %d", nw.Stats.Corrupted, tc.wantCorr)
			}
		})
	}
}

func TestCrashRecoveryKeepsStableState(t *testing.T) {
	r := NewReplica()
	applied, err := r.Apply("ord-1")
	if err != nil || !applied {
		t.Fatalf("first apply = (%v, %v), want (true, nil)", applied, err)
	}

	r.Crash()
	if _, err := r.Apply("ord-1"); !errors.Is(err, ErrCrashed) {
		t.Fatalf("apply while crashed: err = %v, want ErrCrashed", err)
	}
	if r.Applied("ord-1") {
		t.Fatal("volatile state survived crash")
	}

	r.Recover()
	if !r.Applied("ord-1") {
		t.Fatal("stable state missing after recovery")
	}
	applied, err = r.Apply("ord-1")
	if err != nil || applied {
		t.Fatalf("replayed apply = (%v, %v), want (false, nil)", applied, err)
	}
}

=== RUN   TestFaultInjectionTable
=== RUN   TestFaultInjectionTable/clean_link
=== RUN   TestFaultInjectionTable/one_message_lost
=== RUN   TestFaultInjectionTable/partition_blocks_delivery
=== RUN   TestFaultInjectionTable/receiver_crashed
=== RUN   TestFaultInjectionTable/byzantine_payload_rewrite
=== RUN   TestFaultInjectionTable/healed_partition_delivers
--- PASS: TestFaultInjectionTable (0.00s)
    --- PASS: TestFaultInjectionTable/clean_link (0.00s)
    --- PASS: TestFaultInjectionTable/one_message_lost (0.00s)
    --- PASS: TestFaultInjectionTable/partition_blocks_delivery (0.00s)
    --- PASS: TestFaultInjectionTable/receiver_crashed (0.00s)
    --- PASS: TestFaultInjectionTable/byzantine_payload_rewrite (0.00s)
    --- PASS: TestFaultInjectionTable/healed_partition_delivers (0.00s)
=== RUN   TestCrashRecoveryKeepsStableState
--- PASS: TestCrashRecoveryKeepsStableState (0.00s)
PASS
ok  	archlab/internal/platform/faultnet	0.470s