Add local storage fault profile

Author: lemon-mint

MODEL_EQ_REPORT.MD

@@ -26,22 +26,23 @@ This report maps the current finite TLA+ safety model to the current Go implemen
 | `tla/Quorum.tla:9` and `tla/Quorum.tla:55` TLA quorum formulas/intersections | `epaxos/quorum.go:15` Go quorum calculation | Both define supported cluster sizes one through seven, majority slow quorum `n/2 + 1`, conservative fast quorum `n - ((n - 1) / 4)`, explicit expected tables, and quorum-intersection invariants checked by TLC and Go tests. |
 | `tla/EPaxosKVConflict.cfg:11` selective command set | `examples/kv/kv.go:368` Go transaction conflict-key construction | The KV conflict TLC config models independent single-key commands and a transaction-like command that conflicts with both keys; atomic multi-key apply and duplicate-key payload ordering remain implementation-tested at the example storage layer. |
 | Not represented in current TLA safety model | `examples/kv/kv.go:305` Go scan implementation | Advanced scan ordering, reverse ordering, timestamp collision behavior, and Pebble durability are example storage semantics verified by Go tests rather than by the current TLA state. |
-| Not represented in current TLA safety model | `examples/kv/cmd/kvnode/main.go:259` transport fault control and `jepsen/src/moreconsensus/epaxos_test.clj:183` transport nemesis | Local transport partition control is an external validation mechanism for the KV example and Jepsen workload rather than modeled state. |
+| Not represented in current TLA safety model | `examples/kv/cmd/kvnode/main.go:285` transport fault control, `examples/kv/cmd/kvnode/main.go:311` storage fault control, `jepsen/src/moreconsensus/epaxos_test.clj:196` transport nemesis, and `jepsen/src/moreconsensus/epaxos_test.clj:224` storage nemesis | Local transport partition and storage-unavailable controls are external validation mechanisms for the KV example and Jepsen workload rather than modeled state. |
 
 ## Current verification evidence
 
-- Observed `tests/ci.sh` passing after the hardening changes. The gate ran root Go tests, root coverage at 100.0%, example Go tests, example coverage at 100.0%, tagged kvnode tests, root/example/kvnode race checks, TLC model checks, the local Jepsen restart profile, and the repository text audit.
+- Observed `tests/ci.sh` passing after the storage-fault hardening changes. The gate ran root Go tests, root coverage at 100.0%, example Go tests, example coverage at 100.0%, tagged kvnode tests, root/example/kvnode race checks, TLC model checks, local Jepsen restart, transport, and storage profiles, and the repository text audit.
 - Observed `go test ./epaxos` passing with deterministic randomized simulation coverage for three- and five-node clusters, duplicate/drop/reorder delivery, logical ticks, restarts, dependency-vector prefix execution ordering, Ready/Advance durability, allocation/pool ownership assertions, decoder fuzz seeds, and decoder error cleanup.
-- Observed `go test ./examples/kv` and `go test -tags kvnode ./examples/kv/cmd/kvnode` passing with KV duplicate-key transaction payload-order semantics, deduplicated EPAXOS conflict keys, post-Advance executed-record persistence, negative scan-limit rejection, and transport fault-route coverage.
+- Observed `go test ./examples/kv` and `go test -tags kvnode ./examples/kv/cmd/kvnode` passing with KV duplicate-key transaction payload-order semantics, deduplicated EPAXOS conflict keys, post-Advance executed-record persistence, negative scan-limit rejection, transport fault-route coverage, and storage fault-route coverage that rejects work before RawNode progress.
 - Observed `go test ./epaxos -coverprofile=coverage.out -count=1` reporting 100.0% statement coverage and `go tool cover -func=coverage.out` reporting `Advance` at 100.0% after the executed-record acknowledgement cap test.
 - Observed `tests/tla_model_check.sh` running TLC for `tla/EPaxos.cfg` with 11162 generated states and 2002 distinct states, `tla/EPaxosKVConflict.cfg` with 3042650 generated states and 166034 distinct states, `tla/EPaxosThreeReplica.cfg` with 1104121 generated states and 123821 distinct states, and `tla/Quorum.cfg` with 2 generated states and 1 distinct state; all completed with no invariant violation.
-- Observed targeted `lein test moreconsensus.epaxos-test-test` from `jepsen` passing 22 tests with 57 assertions for register delete normalization, indeterminate mutation response classification, transaction body encoding, grouped reads, scan shape checking, restart fault selection, transport fault selection, client/nemesis routing, restart nemesis behavior, and transport nemesis control requests.
-- Observed `JEPSEN_LOCAL_FAULTS=transport tests/jepsen_local.sh` reporting register `:linearizable {:valid? true}`, scan shape `:scan-shape {:valid? true, :checked 17, :bad-count 0}`, transaction `:txn-atomic {:valid? true, :checked 8, :bad-count 0}`, and overall `:valid? true`.
-- Observed `tests/ci.sh` local restart profile reporting register `:linearizable {:valid? true}`, scan shape `:scan-shape {:valid? true, :checked 4, :bad-count 0}`, transaction `:txn-atomic {:valid? true, :checked 2, :bad-count 0}`, and overall `:valid? true`.
+- Observed targeted `lein test moreconsensus.epaxos-test-test` from `jepsen` passing 26 tests with 67 assertions for register delete normalization, indeterminate mutation response classification, transaction body encoding, grouped reads, scan shape checking, restart fault selection, transport fault selection, storage fault selection, client/nemesis routing, restart nemesis behavior, transport nemesis control requests, and storage nemesis control requests.
+- Observed `tests/ci.sh` local transport profile reporting register `:linearizable {:valid? true}`, scan shape `:scan-shape {:valid? true, :checked 11, :bad-count 0}`, transaction `:txn-atomic {:valid? true, :checked 9, :bad-count 0}`, and overall `:valid? true`.
+- Observed `tests/ci.sh` local restart profile reporting register `:linearizable {:valid? true}`, scan shape `:scan-shape {:valid? true, :checked 6, :bad-count 0}`, transaction `:txn-atomic {:valid? true, :checked 2, :bad-count 0}`, and overall `:valid? true`.
+- Observed `tests/ci.sh` local storage profile reporting register `:linearizable {:valid? true}`, scan shape `:scan-shape {:valid? true, :checked 5, :bad-count 0}`, transaction `:txn-atomic {:valid? true, :checked 5, :bad-count 0}`, and overall `:valid? true`.
 - Observed repository text audit with no disallowed external-project name, scaffolding marker, or wall-clock API matches.
 
 ## Open correspondence risks
 
 - The TLA+ model is finite-bounded for TLC, so it demonstrates correspondence for the configured state space rather than unbounded proof.
-- Checksum, wire-decoder tolerance, storage virtualization, recovery response collection, transport fault control, and example KV apply/scan storage semantics are verified by implementation and Jepsen tests but outside the current TLA safety state.
-- Local Jepsen now covers loopback process restart and transport partition profiles, but it still lacks SSH-managed multi-host deployment, disk-fault nemeses, and long-duration operational coverage.
+- Checksum, wire-decoder tolerance, storage virtualization, recovery response collection, transport fault control, storage fault control, and example KV apply/scan storage semantics are verified by implementation and Jepsen tests but outside the current TLA safety state.
+- Local Jepsen now covers loopback process restart, transport partition, and storage-unavailable profiles, but it still lacks SSH-managed multi-host deployment, destructive disk nemeses, and long-duration operational coverage.

examples/kv/cmd/kvnode/main.go

@@ -38,6 +38,7 @@ type service struct {
 	sendq          chan epaxos.Message
 	nextSeq        uint64
 	transportDrops map[transportLink]struct{}
+	storageFailed  bool
 }
 
 type transportLink struct {
@@ -51,6 +52,10 @@ type transportFaultRequest struct {
 	Drop bool             `json:"drop"`
 }
 
+type storageFaultRequest struct {
+	Fail bool `json:"fail"`
+}
+
 func main() {
 	idFlag := flag.Uint64("id", 1, "replica id")
 	listen := flag.String("listen", ":8080", "HTTP listen address")
@@ -81,6 +86,7 @@ func main() {
 	mux.HandleFunc("/scan", s.handleScan)
 	mux.HandleFunc("/epaxos/message", s.handleMessage)
 	mux.HandleFunc("/faults/transport", s.handleTransportFault)
+	mux.HandleFunc("/faults/storage", s.handleStorageFault)
 	mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("ok")) })
 	log.Printf("kvnode %d listening on %s", s.id, *listen)
 	log.Fatal(http.ListenAndServe(*listen, mux))
@@ -117,6 +123,10 @@ func (s *service) handleKV(w http.ResponseWriter, r *http.Request) {
 	key := []byte(keyText)
 	switch r.Method {
 	case http.MethodGet:
+		if s.storageFaultActive() {
+			http.Error(w, "storage fault active", http.StatusServiceUnavailable)
+			return
+		}
 		if err := s.waitForKeys(r.Context(), key); err != nil {
 			http.Error(w, err.Error(), http.StatusServiceUnavailable)
 			return
@@ -132,6 +142,10 @@ func (s *service) handleKV(w http.ResponseWriter, r *http.Request) {
 		}
 		_, _ = w.Write(value)
 	case http.MethodPut:
+		if s.storageFaultActive() {
+			http.Error(w, "storage fault active", http.StatusServiceUnavailable)
+			return
+		}
 		body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusBadRequest)
@@ -143,6 +157,10 @@ func (s *service) handleKV(w http.ResponseWriter, r *http.Request) {
 		}
 		w.WriteHeader(http.StatusNoContent)
 	case http.MethodDelete:
+		if s.storageFaultActive() {
+			http.Error(w, "storage fault active", http.StatusServiceUnavailable)
+			return
+		}
 		if err := s.proposeAndWait(r.Context(), kv.CommandForDelete(uint64(s.id), s.next(), key)); err != nil {
 			http.Error(w, err.Error(), http.StatusServiceUnavailable)
 			return
@@ -181,6 +199,10 @@ func (s *service) handleTxn(w http.ResponseWriter, r *http.Request) {
 		}
 		ops = append(ops, kv.TxnOp{Delete: op.Delete, Key: []byte(op.Key), Value: []byte(op.Value)})
 	}
+	if s.storageFaultActive() {
+		http.Error(w, "storage fault active", http.StatusServiceUnavailable)
+		return
+	}
 	if err := s.proposeAndWait(r.Context(), kv.CommandForTxn(uint64(s.id), s.next(), ops)); err != nil {
 		http.Error(w, err.Error(), http.StatusServiceUnavailable)
 		return
@@ -222,6 +244,10 @@ func (s *service) handleScan(w http.ResponseWriter, r *http.Request) {
 		}
 		reverse = v
 	}
+	if s.storageFaultActive() {
+		http.Error(w, "storage fault active", http.StatusServiceUnavailable)
+		return
+	}
 	if raw := q.Get("barrier"); raw != "" {
 		parts := strings.Split(raw, ",")
 		keys := make([][]byte, 0, len(parts))
@@ -282,6 +308,39 @@ func (s *service) handleTransportFault(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (s *service) handleStorageFault(w http.ResponseWriter, r *http.Request) {
+	switch r.Method {
+	case http.MethodGet:
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(storageFaultRequest{Fail: s.storageFaultActive()})
+	case http.MethodPost:
+		var req storageFaultRequest
+		if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		s.setStorageFault(req.Fail)
+		w.WriteHeader(http.StatusNoContent)
+	case http.MethodDelete:
+		s.setStorageFault(false)
+		w.WriteHeader(http.StatusNoContent)
+	default:
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+	}
+}
+
+func (s *service) setStorageFault(failed bool) {
+	s.faultMu.Lock()
+	defer s.faultMu.Unlock()
+	s.storageFailed = failed
+}
+
+func (s *service) storageFaultActive() bool {
+	s.faultMu.RLock()
+	defer s.faultMu.RUnlock()
+	return s.storageFailed
+}
+
 func (s *service) setTransportDrop(from, to epaxos.ReplicaID, drop bool) {
 	s.faultMu.Lock()
 	defer s.faultMu.Unlock()
@@ -342,12 +401,20 @@ func (s *service) handleMessage(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "transport link dropped", http.StatusConflict)
 		return
 	}
+	if s.storageFaultActive() {
+		http.Error(w, "storage fault active", http.StatusServiceUnavailable)
+		return
+	}
 	s.mu.Lock()
 	err = s.node.Step(msg)
 	out, drainErr := s.drainLocked()
 	s.mu.Unlock()
-	if err != nil || drainErr != nil {
-		http.Error(w, errorsJoin(err, drainErr).Error(), http.StatusBadRequest)
+	if drainErr != nil {
+		http.Error(w, errorsJoin(err, drainErr).Error(), http.StatusServiceUnavailable)
+		return
+	}
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 	s.send(out)
@@ -377,6 +444,9 @@ func (s *service) waitForKeys(ctx context.Context, keys ...[]byte) error {
 }
 
 func (s *service) proposeAndWait(ctx context.Context, cmd epaxos.Command) error {
+	if s.storageFaultActive() {
+		return fmt.Errorf("storage fault active")
+	}
 	s.mu.Lock()
 	ref, err := s.node.Propose(cmd)
 	out, drainErr := s.drainLocked()

examples/kv/cmd/kvnode/main_test.go

@@ -170,7 +170,6 @@ func TestHandleTxnAcceptsDuplicateKeysWithFinalDelete(t *testing.T) {
 	}
 }
 
-
 func TestHandleKVRepeatedPutWaitsForNewAppliedRef(t *testing.T) {
 	s := newTestService(t)
 	for i := 1; i <= 2; i++ {
@@ -302,6 +301,122 @@ func TestHandleMessageRejectsMalformedTransportPayload(t *testing.T) {
 	}
 }
 
+func TestHandleStorageFaultSetsListsAndClearsFailure(t *testing.T) {
+	s := newTestService(t)
+
+	requireStorageFault(t, s, false)
+
+	set := httptest.NewRecorder()
+	s.handleStorageFault(set, httptest.NewRequest(http.MethodPost, "/faults/storage", bytes.NewReader([]byte(`{"fail":true}`))))
+	if set.Code != http.StatusNoContent {
+		t.Fatalf("set status=%d body=%q", set.Code, set.Body.String())
+	}
+	requireStorageFault(t, s, true)
+
+	clearByPost := httptest.NewRecorder()
+	s.handleStorageFault(clearByPost, httptest.NewRequest(http.MethodPost, "/faults/storage", bytes.NewReader([]byte(`{"fail":false}`))))
+	if clearByPost.Code != http.StatusNoContent {
+		t.Fatalf("clear post status=%d body=%q", clearByPost.Code, clearByPost.Body.String())
+	}
+	requireStorageFault(t, s, false)
+
+	setAgain := httptest.NewRecorder()
+	s.handleStorageFault(setAgain, httptest.NewRequest(http.MethodPost, "/faults/storage", bytes.NewReader([]byte(`{"fail":true}`))))
+	if setAgain.Code != http.StatusNoContent {
+		t.Fatalf("set again status=%d body=%q", setAgain.Code, setAgain.Body.String())
+	}
+
+	clearByDelete := httptest.NewRecorder()
+	s.handleStorageFault(clearByDelete, httptest.NewRequest(http.MethodDelete, "/faults/storage", nil))
+	if clearByDelete.Code != http.StatusNoContent {
+		t.Fatalf("delete status=%d body=%q", clearByDelete.Code, clearByDelete.Body.String())
+	}
+	requireStorageFault(t, s, false)
+}
+
+func TestHandleStorageFaultRejectsMalformedRequests(t *testing.T) {
+	tests := []struct {
+		name   string
+		method string
+		body   []byte
+		want   int
+	}{
+		{name: "wrong method", method: http.MethodPut, want: http.StatusMethodNotAllowed},
+		{name: "malformed json", method: http.MethodPost, body: []byte(`{"fail":`), want: http.StatusBadRequest},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := newTestService(t)
+			s.setStorageFault(true)
+			rr := httptest.NewRecorder()
+			s.handleStorageFault(rr, httptest.NewRequest(tc.method, "/faults/storage", bytes.NewReader(tc.body)))
+			if rr.Code != tc.want {
+				t.Fatalf("status=%d body=%q", rr.Code, rr.Body.String())
+			}
+			requireStorageFault(t, s, true)
+		})
+	}
+}
+
+func TestStorageFaultRejectsClientRequestsBeforeConsensusProgress(t *testing.T) {
+	tests := []struct {
+		name   string
+		method string
+		target string
+		body   []byte
+		handle func(*service, http.ResponseWriter, *http.Request)
+	}{
+		{name: "put", method: http.MethodPut, target: "/kv/alpha", body: []byte("one"), handle: (*service).handleKV},
+		{name: "delete", method: http.MethodDelete, target: "/kv/alpha", handle: (*service).handleKV},
+		{name: "get", method: http.MethodGet, target: "/kv/alpha", handle: (*service).handleKV},
+		{name: "txn", method: http.MethodPost, target: "/txn", body: []byte(`[{"key":"alpha","value":"one"}]`), handle: (*service).handleTxn},
+		{name: "scan barrier", method: http.MethodGet, target: "/scan?barrier=alpha", handle: (*service).handleScan},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := newTestService(t)
+			set := httptest.NewRecorder()
+			s.handleStorageFault(set, httptest.NewRequest(http.MethodPost, "/faults/storage", bytes.NewReader([]byte(`{"fail":true}`))))
+			if set.Code != http.StatusNoContent {
+				t.Fatalf("set fault status=%d body=%q", set.Code, set.Body.String())
+			}
+
+			rr := httptest.NewRecorder()
+			tc.handle(s, rr, httptest.NewRequest(tc.method, tc.target, bytes.NewReader(tc.body)))
+			if rr.Code != http.StatusServiceUnavailable || rr.Body.String() != "storage fault active\n" {
+				t.Fatalf("status=%d body=%q", rr.Code, rr.Body.String())
+			}
+			requireNoConsensusProgress(t, s)
+		})
+	}
+}
+
+func TestStorageFaultRejectsInboundMessageBeforeSteppingNode(t *testing.T) {
+	s := newTestClusterService(t, []epaxos.ReplicaID{1, 2})
+	s.setStorageFault(true)
+	msg := epaxos.Message{
+		Type:    epaxos.MsgPreAccept,
+		From:    2,
+		To:      1,
+		Ref:     epaxos.InstanceRef{Replica: 2, Instance: 1, Conf: 1},
+		Ballot:  epaxos.Ballot{Replica: 2},
+		Seq:     1,
+		Deps:    []epaxos.InstanceNum{0, 0},
+		Command: epaxos.Command{ID: epaxos.CommandID{Client: 2, Sequence: 1}, ConflictKeys: [][]byte{[]byte("blocked")}},
+	}
+	buf, err := epaxos.EncodeMessage(nil, msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rr := httptest.NewRecorder()
+	s.handleMessage(rr, httptest.NewRequest(http.MethodPost, "/epaxos/message", bytes.NewReader(buf)))
+	if rr.Code != http.StatusServiceUnavailable || rr.Body.String() != "storage fault active\n" {
+		t.Fatalf("status=%d body=%q", rr.Code, rr.Body.String())
+	}
+	requireNoConsensusProgress(t, s)
+}
+
 func TestHandleTransportFaultSetsListsAndClearsDroppedLinks(t *testing.T) {
 	s := newTestService(t)
 
@@ -422,6 +537,30 @@ func TestHandleMessageDropsConfiguredInboundTransportLinkBeforeSteppingNode(t *t
 	}
 }
 
+func requireStorageFault(t *testing.T, s *service, want bool) {
+	t.Helper()
+	rr := httptest.NewRecorder()
+	s.handleStorageFault(rr, httptest.NewRequest(http.MethodGet, "/faults/storage", nil))
+	if rr.Code != http.StatusOK {
+		t.Fatalf("list status=%d body=%q", rr.Code, rr.Body.String())
+	}
+	var got storageFaultRequest
+	if err := json.Unmarshal(rr.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	if got.Fail != want {
+		t.Fatalf("storage fault active=%t, want %t", got.Fail, want)
+	}
+}
+
+func requireNoConsensusProgress(t *testing.T, s *service) {
+	t.Helper()
+	status := s.node.Status()
+	if len(status.Instances) != 0 || len(status.Executed) != 0 {
+		t.Fatalf("node status after storage fault: instances=%v executed=%v", status.Instances, status.Executed)
+	}
+}
+
 func requireTransportDrops(t *testing.T, body []byte, want []transportFaultRequest) {
 	t.Helper()
 	var got []transportFaultRequest