From fa9e9504ad18635aac38f36b4ef47a1bf2b2fa8c Mon Sep 17 00:00:00 2001
From: Marek Siarkowicz <siarkowicz@google.com>
Date: Sun, 21 Apr 2024 20:49:46 +0200
Subject: [PATCH] Handle watch responses with error

Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>
---
 tests/robustness/model/watch.go            |  1 +
 tests/robustness/traffic/client.go         |  4 +++
 tests/robustness/validate/validate_test.go | 32 ++++++++++++++++++++++
 tests/robustness/validate/watch.go         | 12 ++++----
 4 files changed, 43 insertions(+), 6 deletions(-)

diff --git a/tests/robustness/model/watch.go b/tests/robustness/model/watch.go
index 43bce2f71..fc880e30e 100644
--- a/tests/robustness/model/watch.go
+++ b/tests/robustness/model/watch.go
@@ -26,4 +26,5 @@ type WatchResponse struct {
 	IsProgressNotify bool
 	Revision         int64
 	Time             time.Duration
+	Error            string
 }
diff --git a/tests/robustness/traffic/client.go b/tests/robustness/traffic/client.go
index 94fc3c41e..0fd20e452 100644
--- a/tests/robustness/traffic/client.go
+++ b/tests/robustness/traffic/client.go
@@ -253,6 +253,10 @@ func ToWatchResponse(r clientv3.WatchResponse, baseTime time.Time) model.WatchRe
 	}
 	resp.IsProgressNotify = r.IsProgressNotify()
 	resp.Revision = r.Header.Revision
+	err := r.Err()
+	if err != nil {
+		resp.Error = r.Err().Error()
+	}
 	return resp
 }
 
diff --git a/tests/robustness/validate/validate_test.go b/tests/robustness/validate/validate_test.go
index 6a50fd88a..256b07898 100644
--- a/tests/robustness/validate/validate_test.go
+++ b/tests/robustness/validate/validate_test.go
@@ -1228,6 +1228,38 @@ func TestValidateWatch(t *testing.T) {
 				putPersistedEvent("c", "3", 4, true),
 			},
 		},
+		{
+			name: "Reliable - ignore empty last error response - pass",
+			reports: []report.ClientReport{
+				{
+					Watch: []model.WatchOperation{
+						{
+							Request: model.WatchRequest{
+								WithPrefix: true,
+							},
+							Responses: []model.WatchResponse{
+								{
+									Events: []model.WatchEvent{
+										putWatchEvent("a", "1", 2, true),
+										putWatchEvent("b", "2", 3, true),
+										putWatchEvent("c", "3", 4, true),
+									},
+								},
+								{
+									Revision: 5,
+									Error:    "etcdserver: mvcc: required revision has been compacted",
+								},
+							},
+						},
+					},
+				},
+			},
+			eventHistory: []model.PersistedEvent{
+				putPersistedEvent("a", "1", 2, true),
+				putPersistedEvent("b", "2", 3, true),
+				putPersistedEvent("c", "3", 4, true),
+			},
+		},
 		{
 			name: "Resumable - watch revision from middle event - pass",
 			reports: []report.ClientReport{
diff --git a/tests/robustness/validate/watch.go b/tests/robustness/validate/watch.go
index b42fb6b8a..4b248971f 100644
--- a/tests/robustness/validate/watch.go
+++ b/tests/robustness/validate/watch.go
@@ -307,13 +307,13 @@ func firstExpectedRevision(op model.WatchOperation) int64 {
 }
 
 func lastRevision(op model.WatchOperation) int64 {
-	if len(op.Responses) > 0 {
-		lastResp := op.Responses[len(op.Responses)-1]
-		if lastResp.IsProgressNotify {
-			return lastResp.Revision
+	for i := len(op.Responses) - 1; i >= 0; i-- {
+		resp := op.Responses[i]
+		if resp.IsProgressNotify {
+			return resp.Revision
 		}
-		if len(lastResp.Events) > 0 {
-			lastEvent := lastResp.Events[len(lastResp.Events)-1]
+		if len(resp.Events) > 0 {
+			lastEvent := resp.Events[len(resp.Events)-1]
 			return lastEvent.Revision
 		}
 	}