Avoid duplicate inbound persistence

Author: lemon-mint

epaxos/node.go

@@ -557,7 +557,8 @@ func (n *RawNode) handlePreAccept(m Message) {
 	attrs := n.computeAttrs(m.Command, m.Ref)
 	attrs = mergeAttrs(attrs, m.Attributes())
 	rec := InstanceRecord{Ref: m.Ref, Ballot: m.Ballot, Status: StatusPreAccepted, Seq: attrs.Seq, Deps: attrs.Deps, Command: inboundCommand(m.Command)}
-	if old := n.instances[m.Ref]; old != nil {
+	old := n.instances[m.Ref]
+	if old != nil {
 		if old.rec.Status >= StatusCommitted {
 			n.sendCommitTo(m.From, old.rec)
 			return
@@ -568,6 +569,11 @@ func (n *RawNode) handlePreAccept(m Message) {
 		}
 	}
 	rec.Checksum = ChecksumRecord(rec)
+	if old != nil && old.rec.Status == StatusPreAccepted && instanceRecordEqual(old.rec, rec) {
+		resp := Message{Type: MsgPreAcceptResp, From: n.id, To: m.From, Ref: m.Ref, Ballot: rec.Ballot, Seq: rec.Seq, Deps: rec.Deps, RecordStatus: rec.Status}
+		n.enqueueMessage(resp)
+		return
+	}
 	n.instances[m.Ref] = &instance{rec: rec, phase: phasePreAccept}
 	n.indexConflicts(rec)
 	n.enqueueRecord(rec)
@@ -610,7 +616,8 @@ func (n *RawNode) handlePreAcceptResp(m Message) {
 }
 
 func (n *RawNode) handleAccept(m Message) {
-	if old := n.instances[m.Ref]; old != nil {
+	old := n.instances[m.Ref]
+	if old != nil {
 		if old.rec.Status >= StatusCommitted {
 			n.sendCommitTo(m.From, old.rec)
 			return
@@ -622,6 +629,11 @@ func (n *RawNode) handleAccept(m Message) {
 	}
 	rec := InstanceRecord{Ref: m.Ref, Ballot: m.Ballot, Status: StatusAccepted, Seq: m.Seq, Deps: append([]InstanceNum(nil), m.Deps...), Command: inboundCommand(m.Command)}
 	rec.Checksum = ChecksumRecord(rec)
+	if old != nil && old.rec.Status == StatusAccepted && instanceRecordEqual(old.rec, rec) {
+		resp := Message{Type: MsgAcceptResp, From: n.id, To: m.From, Ref: m.Ref, Ballot: rec.Ballot, Seq: rec.Seq, Deps: rec.Deps, RecordStatus: rec.Status}
+		n.enqueueMessage(resp)
+		return
+	}
 	n.instances[m.Ref] = &instance{rec: rec, phase: phaseAccept}
 	n.indexConflicts(rec)
 	n.enqueueRecord(rec)

epaxos/sim_test.go

@@ -158,6 +158,81 @@ func TestDuplicateMessagesAndMalformedInput(t *testing.T) {
 	}
 }
 
+func TestDuplicateInboundPreAcceptAndAcceptDoNotQueueDuplicateRecords(t *testing.T) {
+	ref := InstanceRef{Replica: 1, Instance: 1, Conf: 1}
+	command := Command{ID: CommandID{Client: 10, Sequence: 20}, Payload: []byte("duplicate-inbound"), ConflictKeys: [][]byte{[]byte("duplicate-key")}}
+	tests := []struct {
+		name       string
+		msg        Message
+		wantStatus Status
+	}{
+		{
+			name: "preaccept",
+			msg: Message{
+				Type:    MsgPreAccept,
+				From:    1,
+				To:      2,
+				Ref:     ref,
+				Ballot:  Ballot{Replica: 1},
+				Seq:     1,
+				Deps:    []InstanceNum{0, 0, 0},
+				Command: command,
+			},
+			wantStatus: StatusPreAccepted,
+		},
+		{
+			name: "accept",
+			msg: Message{
+				Type:    MsgAccept,
+				From:    1,
+				To:      2,
+				Ref:     ref,
+				Ballot:  Ballot{Replica: 1},
+				Seq:     1,
+				Deps:    []InstanceNum{0, 0, 0},
+				Command: command,
+			},
+			wantStatus: StatusAccepted,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			store := NewMemoryStorage()
+			rn, err := NewRawNode(Config{ID: 2, Voters: makeIDs(3), Storage: store})
+			if err != nil {
+				t.Fatal(err)
+			}
+			tt.msg.Checksum = ChecksumMessage(tt.msg)
+
+			if err := rn.Step(tt.msg); err != nil {
+				t.Fatalf("first Step(%s) err=%v", tt.msg.Type, err)
+			}
+			first := rn.Ready()
+			if len(first.Records) != 1 || first.Records[0].Ref != ref || first.Records[0].Status != tt.wantStatus {
+				t.Fatalf("first duplicate target ready records = %#v, want exactly one %s record for %s", first.Records, tt.wantStatus, ref)
+			}
+			if err := store.ApplyReady(first); err != nil {
+				t.Fatal(err)
+			}
+			advanceOK(t, rn, first)
+
+			if err := rn.Step(tt.msg); err != nil {
+				t.Fatalf("duplicate Step(%s) err=%v", tt.msg.Type, err)
+			}
+			duplicate := rn.Ready()
+			if len(duplicate.Records) != 0 {
+				t.Fatalf("duplicate %s ready records = %#v, want no durable records for unchanged %s", tt.msg.Type, duplicate.Records, ref)
+			}
+			if !duplicate.Empty() {
+				if err := store.ApplyReady(duplicate); err != nil {
+					t.Fatal(err)
+				}
+				advanceOK(t, rn, duplicate)
+			}
+		})
+	}
+}
+
 func TestCodecChecksumZeroCopy(t *testing.T) {
 	m := Message{
 		Type:         MsgCommit,