Close completion audit gaps

Author: lemon-mint

.gitignore

@@ -29,6 +29,7 @@ states/
 jepsen/target/
 jepsen/.lein-failures
 jepsen/store/
+kvnode-data/
 
 # env file
 .env

MODEL_EQ_REPORT.MD

@@ -7,24 +7,24 @@ This report maps the current finite TLA+ safety model to the current Go implemen
 | Model anchor | Implementation anchor | Evidence |
 | --- | --- | --- |
 | `tla/EPaxos.tla:27` TLA Init | `epaxos/node.go:98` Go node construction | Initialization sets empty instance state before transport steps; Go also restores durable state and configuration history before accepting input. |
-| `tla/EPaxos.tla:15` TLA SafeDeps | `epaxos/node.go:702` Go attribute computation | Both model and implementation require known conflicting commands to appear in dependency attributes. |
+| `tla/EPaxos.tla:15` TLA SafeDeps | `epaxos/node.go:706` Go attribute computation | Both model and implementation require known conflicting commands to appear in dependency attributes. |
 | `tla/EPaxos.tla:37` TLA PreAccept | `epaxos/node.go:257` Go proposal path | Local proposal computes attributes, persists pre-accepted records, indexes conflicts, and emits pre-accept messages. |
-| `tla/EPaxos.tla:37` TLA PreAccept receive | `epaxos/node.go:401` Go pre-accept handler | Remote pre-accept merges local conflicts with proposer attributes and stores the pre-accepted value. |
-| `tla/EPaxos.tla:49` TLA Accept | `epaxos/node.go:457` Go accept handler | Accept stores slow-path attributes unless the instance is already committed or promised higher. |
-| `tla/EPaxos.tla:58` TLA Commit | `epaxos/node.go:501` Go commit handler | Commit stores final value/attributes and invokes dependency-closed execution. |
-| `tla/EPaxos.tla:64` TLA Prepare | `epaxos/node.go:514` Go prepare handler | Prepare persists higher promises and returns local accepted/committed state for recovery; TLA abstracts this as the ballot-raising recovery step. |
-| Not represented in current TLA safety model | `epaxos/node.go:540` Go prepare response handler | Recovery response quorum handling is verified by Go tests, but the current TLA model abstracts away recovery quorum collection. |
-| `tla/EPaxos.tla:71` TLA Execute | `epaxos/node.go:802` Go executor | Execution collapses committed dependency SCCs, persists executed status, applies configuration commands, and emits application commands. |
-| `tla/EPaxos.tla:91` TLA DependencyClosure | `epaxos/node.go:900` Go component readiness | Component readiness requires outside dependencies to be executed or absent before application emission. |
-| `tla/EPaxos.tla:93` TLA ConflictOrder | `epaxos/node.go:949` Go dependency references | Both require conflicting executed commands to be ordered by dependencies; Go interprets dependency vectors through configuration history. |
+| `tla/EPaxos.tla:37` TLA PreAccept receive | `epaxos/node.go:405` Go pre-accept handler | Remote pre-accept merges local conflicts with proposer attributes and stores the pre-accepted value. |
+| `tla/EPaxos.tla:49` TLA Accept | `epaxos/node.go:461` Go accept handler | Accept stores slow-path attributes unless the instance is already committed or promised higher. |
+| `tla/EPaxos.tla:58` TLA Commit | `epaxos/node.go:505` Go commit handler | Commit stores final value/attributes and invokes dependency-closed execution. |
+| `tla/EPaxos.tla:64` TLA Prepare | `epaxos/node.go:518` Go prepare handler | Prepare persists higher promises and returns local accepted/committed state for recovery; TLA abstracts this as the ballot-raising recovery step. |
+| Not represented in current TLA safety model | `epaxos/node.go:544` Go prepare response handler | Recovery response quorum handling is verified by Go tests, but the current TLA model abstracts away recovery quorum collection. |
+| `tla/EPaxos.tla:71` TLA Execute | `epaxos/node.go:813` Go executor | Execution collapses committed dependency SCCs, persists executed status, applies configuration commands, and emits application commands. |
+| `tla/EPaxos.tla:91` TLA DependencyClosure | `epaxos/node.go:911` Go component readiness | Component readiness requires outside dependencies to be executed or absent before application emission. |
+| `tla/EPaxos.tla:93` TLA ConflictOrder | `epaxos/node.go:960` Go dependency references | Both require conflicting executed commands to be ordered by dependencies; Go interprets dependency vectors through configuration history. |
 | `tla/EPaxos.tla:78` TLA Tick | `epaxos/node.go:237` Go logical tick | Tick advances logical time and dispatches logical timers; no wall-clock API is used in the core. |
 | Not represented in current TLA safety model | `epaxos/checksum.go:65` Go record checksum | Durable record checksum coverage is implementation validation rather than modeled state. |
 | Not represented in current TLA safety model | `epaxos/checksum.go:84` Go message checksum | Transport checksum coverage is implementation validation rather than modeled state. |
 | Not represented in current TLA safety model | `epaxos/codec.go:40` Go decoder | Wire-decoder input tolerance is implementation validation rather than modeled state. |
 | Not represented in current TLA safety model | `epaxos/storage.go:9` Go storage interface | Storage virtualization is an implementation boundary used by deterministic tests rather than modeled state. |
 | Not represented in current TLA safety model | `epaxos/quorum.go:15` Go quorum calculation | Quorum arithmetic for cluster sizes one through seven is verified by Go tests; the current TLA constants enumerate voters directly. |
-| Not represented in current TLA safety model | `examples/kv/kv.go:102` Go transaction batch apply | Example KV transactions are opaque EPaxos commands at the model boundary; atomic multi-key apply is verified by Go tests rather than by the current TLA state. |
-| Not represented in current TLA safety model | `examples/kv/kv.go:169` 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/kv.go:168` Go transaction batch apply | Example KV transactions are opaque EPaxos commands at the model boundary; atomic multi-key apply is verified by Go tests rather than by the current TLA state. |
+| Not represented in current TLA safety model | `examples/kv/kv.go:256` 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. |
 
 ## Current verification evidence
 
@@ -33,10 +33,10 @@ This report maps the current finite TLA+ safety model to the current Go implemen
 - Observed `go test ./...` from `examples/kv` passing.
 - Observed `go test -coverprofile=coverage.out ./...` from `examples/kv` plus `go tool cover -func=coverage.out` reporting 100.0% statement coverage for the example module.
 - Observed `go test -tags kvnode ./cmd/kvnode` from `examples/kv` passing for the tagged HTTP service package.
-- Observed `go test ./...` coverage including cluster destruction and restart via `TestRestartAllRawNodesRetainsExecutedAndAppliesOnlyNewCommand`, remove-voter progress via `TestRemoveVoterConfChangeAllowsLaterProgress`, storage write retry via `TestWriteErrorKeepsReadyForRetry`, partition heal convergence via `TestFiveNodePartitionHealConverges`, equal-sequence execution ordering via `TestExecutionEqualSeqTieBreaksByRef`, and inactive dependency filtering via `TestExecutionComponentsSkipInactiveDependencyRefs`.
+- Observed `go test ./...` coverage including deterministic ordered simulation scheduling, cluster destruction and restart via `TestRestartAllRawNodesRetainsExecutedAndAppliesOnlyNewCommand`, remove-voter progress via `TestRemoveVoterConfChangeAllowsLaterProgress`, storage write retry via `TestWriteErrorKeepsReadyForRetry`, partition heal convergence via `TestFiveNodePartitionHealConverges`, equal-sequence execution ordering via `TestExecutionEqualSeqTieBreaksByRef`, inactive dependency filtering via `TestExecutionComponentsSkipInactiveDependencyRefs`, `MaxReadyMessages` batching, and logical-time optimization fast-wait behavior.
 - Observed `go test ./...` coverage in `examples/kv` including transaction atomicity, malformed transaction rejection, transaction batch error propagation, transaction commit error propagation, apply-order visible-version selection, timestamp collision overwrite behavior, ApplyCommitted close/open durability, delete-tombstone scan suppression, Pebble close/open durability, and reverse scans returning the newest version for repeated keys.
 - Observed TLC passing with `/opt/homebrew/opt/openjdk/bin/java -cp /tmp/tla2tools.jar tlc2.TLC -config tla/EPaxos.cfg tla/EPaxos.tla`, generating 11162 states, 2002 distinct states, and no invariant violation.
-- Observed `lein test` from `jepsen` passing namespace and dependency loading, and observed `tests/ci.sh` running a local three-node Jepsen workload against deployed `kvnode` processes with `lein run test --no-ssh --nodes 127.0.0.1:19081,127.0.0.1:19082,127.0.0.1:19083 --time-limit 5 --concurrency 3` reporting register `:linearizable {:valid? true}` and transaction `:txn-atomic {:valid? true, :checked 27, :bad-count 0}`.
+- Observed `lein test` from `jepsen` passing namespace and dependency loading, and observed `tests/ci.sh` running a local three-node Jepsen workload against deployed `kvnode` processes with `lein run test --no-ssh --nodes 127.0.0.1:19081,127.0.0.1:19082,127.0.0.1:19083 --time-limit 5 --concurrency 3` reporting register `:linearizable {:valid? true}` and transaction `:txn-atomic {:valid? true, :checked 23, :bad-count 0}`.
 - Observed repository text audit with no disallowed external-project name, scaffolding marker, or wall-clock API matches.
 
 ## Open correspondence risks

epaxos/node.go

@@ -345,8 +345,12 @@ func (n *RawNode) Ready() Ready {
 	for i := range n.pendingReady.Records {
 		rd.Records[i] = n.pendingReady.Records[i].Clone()
 	}
-	rd.Messages = make([]Message, len(n.pendingReady.Messages))
-	for i := range n.pendingReady.Messages {
+	messageCount := len(n.pendingReady.Messages)
+	if n.maxReadyMessages > 0 && messageCount > n.maxReadyMessages {
+		messageCount = n.maxReadyMessages
+	}
+	rd.Messages = make([]Message, messageCount)
+	for i := range rd.Messages {
 		rd.Messages[i] = n.pendingReady.Messages[i].Clone()
 	}
 	rd.Committed = make([]CommittedCommand, len(n.pendingReady.Committed))

epaxos/remaining_test.go

@@ -72,6 +72,133 @@ func TestRemainingReadyAndProposalBranches(t *testing.T) {
 	zr.schedule(inst, timerAccept, 0)
 }
 
+func TestMaxReadyMessagesCapsOnlyMessages(t *testing.T) {
+	rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3), MaxReadyMessages: 1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	first := InstanceRef{Replica: 1, Instance: 1, Conf: 1}
+	second := InstanceRef{Replica: 1, Instance: 2, Conf: 1}
+	rn.pendingReady = Ready{
+		Records: []InstanceRecord{
+			{Ref: first, Command: Command{Payload: []byte("record-1")}},
+			{Ref: second, Command: Command{Payload: []byte("record-2")}},
+		},
+		Messages: []Message{
+			{Type: MsgPreAccept, From: 1, To: 2, Ref: first, Deps: []InstanceNum{0, 0, 0}},
+			{Type: MsgPreAccept, From: 1, To: 3, Ref: second, Deps: []InstanceNum{0, 0, 0}},
+		},
+		Committed: []CommittedCommand{
+			{Ref: first, Command: Command{Payload: []byte("commit-1")}},
+			{Ref: second, Command: Command{Payload: []byte("commit-2")}},
+		},
+		MustSync: true,
+	}
+
+	rd := rn.Ready()
+	if len(rd.Messages) != 1 || rd.Messages[0].To != 2 {
+		t.Fatalf("first ready messages = %#v, want only the first capped message", rd.Messages)
+	}
+	if len(rd.Records) != 2 || rd.Records[0].Ref != first || rd.Records[1].Ref != second {
+		t.Fatalf("records were capped with messages: %#v", rd.Records)
+	}
+	if len(rd.Committed) != 2 || rd.Committed[0].Ref != first || rd.Committed[1].Ref != second {
+		t.Fatalf("committed commands were capped with messages: %#v", rd.Committed)
+	}
+
+	rn.Advance(rd)
+	tail := rn.Ready()
+	if len(tail.Records) != 0 || len(tail.Committed) != 0 {
+		t.Fatalf("advanced records/committed reappeared with message tail: records=%#v committed=%#v", tail.Records, tail.Committed)
+	}
+	if len(tail.Messages) != 1 || tail.Messages[0].To != 3 || tail.Messages[0].Ref != second {
+		t.Fatalf("message tail ready = %#v, want the unsent second message", tail.Messages)
+	}
+}
+
+func TestTimeOptimizationDelaysSlowAcceptUntilFastWaitTick(t *testing.T) {
+	store := NewMemoryStorage()
+	rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(5), Storage: store, RetryTicks: 10, TimeOptimization: true, TimeOptimizationTicks: 3})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ref, err := rn.Propose(Command{ID: CommandID{Client: 7, Sequence: 1}, Payload: []byte("value"), ConflictKeys: [][]byte{[]byte("key")}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rd := rn.Ready()
+	if len(rd.Messages) != 4 {
+		t.Fatalf("initial preaccept messages = %#v", rd.Messages)
+	}
+	if err := store.ApplyReady(rd); err != nil {
+		t.Fatal(err)
+	}
+	rn.Advance(rd)
+
+	inst := rn.instances[ref]
+	if inst == nil {
+		t.Fatalf("missing local instance %s", ref)
+	}
+	for _, from := range []ReplicaID{2, 3} {
+		resp := Message{
+			Type:         MsgPreAcceptResp,
+			From:         from,
+			To:           1,
+			Ref:          ref,
+			Ballot:       inst.rec.Ballot,
+			Seq:          inst.rec.Seq,
+			Deps:         append([]InstanceNum(nil), inst.rec.Deps...),
+			RecordStatus: StatusPreAccepted,
+		}
+		if err := rn.Step(resp); err != nil {
+			t.Fatalf("step preaccept response from %d: %v", from, err)
+		}
+	}
+	if got, slow, fast := len(inst.preOK), rn.q.slowQuorum(), rn.q.fastQuorum(); got != slow || got >= fast {
+		t.Fatalf("preaccept votes = %d, want slow quorum %d below fast quorum %d", got, slow, fast)
+	}
+	if inst.phase != phasePreAccept {
+		t.Fatalf("phase after slow quorum preaccept responses = %d, want preaccept", inst.phase)
+	}
+	if rn.HasReady() {
+		t.Fatalf("slow quorum response produced ready work before fast-wait deadline: %#v", rn.Ready())
+	}
+
+	for tick := uint64(1); tick < 3; tick++ {
+		rn.Tick()
+		if inst.phase != phasePreAccept {
+			t.Fatalf("phase after tick %d = %d, want preaccept before fast-wait deadline", tick, inst.phase)
+		}
+		if rn.HasReady() {
+			t.Fatalf("ready work appeared after tick %d before fast-wait deadline: %#v", tick, rn.Ready())
+		}
+	}
+
+	rn.Tick()
+	if inst.phase != phaseAccept {
+		t.Fatalf("phase at fast-wait deadline = %d, want accept", inst.phase)
+	}
+	rd = rn.Ready()
+	if len(rd.Records) != 1 || rd.Records[0].Ref != ref || rd.Records[0].Status != StatusAccepted {
+		t.Fatalf("fast-wait ready records = %#v, want accepted record for %s", rd.Records, ref)
+	}
+	if len(rd.Messages) != 4 {
+		t.Fatalf("fast-wait accept messages = %#v", rd.Messages)
+	}
+	seen := make(map[ReplicaID]bool, 4)
+	for _, msg := range rd.Messages {
+		if msg.Type != MsgAccept || msg.From != 1 || msg.Ref != ref {
+			t.Fatalf("fast-wait message = %#v, want accept for %s from replica 1", msg, ref)
+		}
+		seen[msg.To] = true
+	}
+	for _, to := range []ReplicaID{2, 3, 4, 5} {
+		if !seen[to] {
+			t.Fatalf("missing accept message to replica %d in %#v", to, rd.Messages)
+		}
+	}
+}
+
 func TestRemainingResponseBranches(t *testing.T) {
 	rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
 	if err != nil {

epaxos/sim_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"sort"
 	"testing"
 )
 
@@ -31,10 +32,20 @@ func newSimCluster(t *testing.T, n int, opt bool) *simCluster {
 	return s
 }
 
+func (s *simCluster) ids() []ReplicaID {
+	ids := make([]ReplicaID, 0, len(s.nodes))
+	for id := range s.nodes {
+		ids = append(ids, id)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	return ids
+}
+
 func (s *simCluster) drain() {
 	for round := 0; round < 1000; round++ {
 		progress := false
-		for id, rn := range s.nodes {
+		for _, id := range s.ids() {
+			rn := s.nodes[id]
 			if !rn.HasReady() {
 				continue
 			}
@@ -67,8 +78,8 @@ func (s *simCluster) drain() {
 
 func (s *simCluster) tickAll(n int) {
 	for i := 0; i < n; i++ {
-		for _, rn := range s.nodes {
-			rn.Tick()
+		for _, id := range s.ids() {
+			s.nodes[id].Tick()
 		}
 		s.drain()
 	}
@@ -83,7 +94,7 @@ func TestClusterSizesOneThroughSevenCommit(t *testing.T) {
 				t.Fatal(err)
 			}
 			s.drain()
-			for id := range s.nodes {
+			for _, id := range s.ids() {
 				if got := len(s.apps[id]); got != 1 {
 					t.Fatalf("node %d applied %d commands", id, got)
 				}
@@ -105,7 +116,7 @@ func TestConflictingConcurrentCommandsConverge(t *testing.T) {
 	}
 	s.drain()
 	want := refs(s.apps[1])
-	for id := range s.nodes {
+	for _, id := range s.ids() {
 		if got := refs(s.apps[id]); fmt.Sprint(got) != fmt.Sprint(want) {
 			t.Fatalf("node %d order %v want %v", id, got, want)
 		}
@@ -190,7 +201,7 @@ func TestRestartAllRawNodesRetainsExecutedAndAppliesOnlyNewCommand(t *testing.T)
 		t.Fatalf("node 1 applied %d commands before restart", got)
 	}
 	firstRef := s.apps[1][0].Ref
-	for id := range s.nodes {
+	for _, id := range s.ids() {
 		if got := len(s.apps[id]); got != 1 {
 			t.Fatalf("node %d applied %d commands before restart", id, got)
 		}
@@ -213,7 +224,8 @@ func TestRestartAllRawNodesRetainsExecutedAndAppliesOnlyNewCommand(t *testing.T)
 		t.Fatal(err)
 	}
 	s.drain()
-	for id, rn := range s.nodes {
+	for _, id := range s.ids() {
+		rn := s.nodes[id]
 		applied := s.apps[id]
 		if len(applied) != 1 {
 			t.Fatalf("node %d applied %d commands after restart: %#v", id, len(applied), applied)
@@ -483,7 +495,7 @@ func TestFiveNodePartitionHealConverges(t *testing.T) {
 	s.drop = map[[2]ReplicaID]bool{}
 	s.tickAll(6)
 	want := refs(s.apps[1])
-	for id := range s.nodes {
+	for _, id := range s.ids() {
 		if got := refs(s.apps[id]); fmt.Sprint(got) != fmt.Sprint(want) {
 			t.Fatalf("node %d refs = %v, want %v", id, got, want)
 		}

examples/kv/kv.go

@@ -67,6 +67,7 @@ type ScanOptions struct {
 	Reverse bool
 }
 
+// Open opens a Pebble-backed key-value store and resumes automatic version timestamps from existing records.
 func Open(path string) (*DB, error) {
 	return open(path, nil)
 }