From a75e3d9051b2e8cb3c6efc1306920d2b7ca62b22 Mon Sep 17 00:00:00 2001
From: Aleksandr Zagaevskiy <a.s.zagaevskiy@gmail.com>
Date: Thu, 14 May 2020 22:25:04 +0300
Subject: [PATCH] Preserve ModifyIndex for unchanged entry in KVS TXN (#7832)

---
 agent/consul/state/kvs.go      |  10 ++-
 agent/consul/state/txn_test.go | 129 +++++++++++++++++++++++++++++++++
 2 files changed, 136 insertions(+), 3 deletions(-)

diff --git a/agent/consul/state/kvs.go b/agent/consul/state/kvs.go
index 61f40897a9..ab412c133b 100644
--- a/agent/consul/state/kvs.go
+++ b/agent/consul/state/kvs.go
@@ -127,13 +127,12 @@ func (s *Store) kvsSetTxn(tx *memdb.Txn, idx uint64, entry *structs.DirEntry, up
 	}
 	existing, _ := existingNode.(*structs.DirEntry)
 
-	// Set the indexes.
+	// Set the CreateIndex.
 	if existing != nil {
 		entry.CreateIndex = existing.CreateIndex
 	} else {
 		entry.CreateIndex = idx
 	}
-	entry.ModifyIndex = idx
 
 	// Preserve the existing session unless told otherwise. The "existing"
 	// session for a new entry is "no session".
@@ -145,10 +144,15 @@ func (s *Store) kvsSetTxn(tx *memdb.Txn, idx uint64, entry *structs.DirEntry, up
 		}
 	}
 
-	// skip write if the entry did not change
+	// Set the ModifyIndex.
 	if existing != nil && existing.Equal(entry) {
+		// Skip further writing in the state store if the entry is not actually
+		// changed. Nevertheless, the input's ModifyIndex should be reset
+		// since the TXN API returns a copy in the response.
+		entry.ModifyIndex = existing.ModifyIndex
 		return nil
 	}
+	entry.ModifyIndex = idx
 
 	// Store the kv pair in the state store and update the index.
 	if err := s.insertKVTxn(tx, entry, false); err != nil {
diff --git a/agent/consul/state/txn_test.go b/agent/consul/state/txn_test.go
index 6a88c7b743..b53f87ce07 100644
--- a/agent/consul/state/txn_test.go
+++ b/agent/consul/state/txn_test.go
@@ -1252,3 +1252,132 @@ func TestStateStore_Txn_KVS_RO_Safety(t *testing.T) {
 		}
 	}
 }
+
+func TestStateStore_Txn_KVS_ModifyIndexes(t *testing.T) {
+	s := testStateStore(t)
+
+	// Create KV entries in the state store.
+	testSetKey(t, s, 1, "foo/a", "bar", nil)
+	testSetKey(t, s, 2, "foo/b", "bar", nil)
+
+	// Set up a transaction that actually changes `a`,
+	// but passes original value for `b`.
+	ops := structs.TxnOps{
+		&structs.TxnOp{
+			KV: &structs.TxnKVOp{
+				Verb: api.KVCAS,
+				DirEnt: structs.DirEntry{
+					Key:   "foo/a",
+					Value: []byte("new"),
+					RaftIndex: structs.RaftIndex{
+						ModifyIndex: 1,
+					},
+				},
+			},
+		},
+		&structs.TxnOp{
+			KV: &structs.TxnKVOp{
+				Verb: api.KVCAS,
+				DirEnt: structs.DirEntry{
+					Key:   "foo/b",
+					Value: []byte("bar"),
+					RaftIndex: structs.RaftIndex{
+						ModifyIndex: 2,
+					},
+				},
+			},
+		},
+	}
+	results, errors := s.TxnRW(3, ops)
+	if len(errors) > 0 {
+		t.Fatalf("err: %v", errors)
+	}
+
+	// Make sure the response looks as expected.
+	expected := structs.TxnResults{
+		&structs.TxnResult{
+			KV: &structs.DirEntry{
+				Key: "foo/a",
+				RaftIndex: structs.RaftIndex{
+					CreateIndex: 1,
+					ModifyIndex: 3,
+				},
+			},
+		},
+		&structs.TxnResult{
+			KV: &structs.DirEntry{
+				Key: "foo/b",
+				RaftIndex: structs.RaftIndex{
+					CreateIndex: 2,
+					ModifyIndex: 2,
+				},
+			},
+		},
+	}
+	if len(results) != len(expected) {
+		t.Fatalf("bad: %v", results)
+	}
+	for i, e := range expected {
+		if e.KV.Key != results[i].KV.Key {
+			t.Fatalf("expected key %s, got %s", e.KV.Key, results[i].KV.Key)
+		}
+		if e.KV.LockIndex != results[i].KV.LockIndex {
+			t.Fatalf("expected lock index %d, got %d", e.KV.LockIndex, results[i].KV.LockIndex)
+		}
+		if e.KV.CreateIndex != results[i].KV.CreateIndex {
+			t.Fatalf("expected create index %d, got %d", e.KV.CreateIndex, results[i].KV.CreateIndex)
+		}
+		if e.KV.ModifyIndex != results[i].KV.ModifyIndex {
+			t.Fatalf("expected modify index %d, got %d", e.KV.ModifyIndex, results[i].KV.ModifyIndex)
+		}
+	}
+
+	// Pull the resulting state store contents.
+	idx, actual, err := s.KVSList(nil, "", nil)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if idx != 3 {
+		t.Fatalf("bad index: %d", idx)
+	}
+
+	// Make sure it looks as expected.
+	entries := structs.DirEntries{
+		&structs.DirEntry{
+			Key:   "foo/a",
+			Value: []byte("new"),
+			RaftIndex: structs.RaftIndex{
+				CreateIndex: 1,
+				ModifyIndex: 3,
+			},
+		},
+		&structs.DirEntry{
+			Key:   "foo/b",
+			Value: []byte("bar"),
+			RaftIndex: structs.RaftIndex{
+				CreateIndex: 2,
+				ModifyIndex: 2,
+			},
+		},
+	}
+	if len(actual) != len(entries) {
+		t.Fatalf("bad len: %d != %d", len(actual), len(entries))
+	}
+	for i, e := range entries {
+		if e.Key != actual[i].Key {
+			t.Fatalf("expected key %s, got %s", e.Key, actual[i].Key)
+		}
+		if string(e.Value) != string(actual[i].Value) {
+			t.Fatalf("expected value %s, got %s", e.Value, actual[i].Value)
+		}
+		if e.LockIndex != actual[i].LockIndex {
+			t.Fatalf("expected lock index %d, got %d", e.LockIndex, actual[i].LockIndex)
+		}
+		if e.CreateIndex != actual[i].CreateIndex {
+			t.Fatalf("expected create index %d, got %d", e.CreateIndex, actual[i].CreateIndex)
+		}
+		if e.ModifyIndex != actual[i].ModifyIndex {
+			t.Fatalf("expected modify index %d, got %d", e.ModifyIndex, actual[i].ModifyIndex)
+		}
+	}
+}