Add tests to serializable operations validation

Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>

tests/robustness/validate/operations.go

@@ -16,9 +16,6 @@ package validate
 
 import (
 	"fmt"
-	"reflect"
-	"sort"
-	"testing"
 	"time"
 
 	"github.com/anishathalye/porcupine"
@@ -26,6 +23,7 @@ import (
 	"go.uber.org/zap"
 
 	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.etcd.io/etcd/tests/v3/robustness/report"
 )
 
 func validateLinearizableOperationsAndVisualize(lg *zap.Logger, operations []porcupine.Operation, timeout time.Duration) (result porcupine.CheckResult, visualize func(basepath string) error) {
@@ -52,45 +50,50 @@ func validateLinearizableOperationsAndVisualize(lg *zap.Logger, operations []por
 	}
 }
 
-func validateSerializableOperations(t *testing.T, lg *zap.Logger, operations []porcupine.Operation, persistedRequests []model.EtcdRequest) {
+func validateSerializableOperations(lg *zap.Logger, operations []porcupine.Operation, replay *model.EtcdReplay) (lastErr error) {
 	lg.Info("Validating serializable operations")
-	staleReads := filterSerializableReads(operations)
-	if len(staleReads) == 0 {
-		return
-	}
-	sort.Slice(staleReads, func(i, j int) bool {
-		return staleReads[i].Input.(model.EtcdRequest).Range.Revision < staleReads[j].Input.(model.EtcdRequest).Range.Revision
-	})
-	replay := model.NewReplay(persistedRequests)
-	for _, read := range staleReads {
+	for _, read := range operations {
 		request := read.Input.(model.EtcdRequest)
 		response := read.Output.(model.MaybeEtcdResponse)
-		validateSerializableOperation(t, replay, request, response)
+		err := validateSerializableRead(lg, replay, request, response)
+		if err != nil {
+			lastErr = err
+		}
 	}
+	return lastErr
 }
 
-func filterSerializableReads(operations []porcupine.Operation) []porcupine.Operation {
+func filterSerializableOperations(clients []report.ClientReport) []porcupine.Operation {
 	resp := []porcupine.Operation{}
-	for _, op := range operations {
-		request := op.Input.(model.EtcdRequest)
-		if request.Type == model.Range && request.Range.Revision != 0 {
-			resp = append(resp, op)
+	for _, client := range clients {
+		for _, op := range client.KeyValue {
+			request := op.Input.(model.EtcdRequest)
+			if request.Type == model.Range && request.Range.Revision != 0 {
+				resp = append(resp, op)
+			}
 		}
 	}
 	return resp
 }
 
-func validateSerializableOperation(t *testing.T, replay *model.EtcdReplay, request model.EtcdRequest, response model.MaybeEtcdResponse) {
+func validateSerializableRead(lg *zap.Logger, replay *model.EtcdReplay, request model.EtcdRequest, response model.MaybeEtcdResponse) error {
 	if response.PartialResponse || response.Error != "" {
-		return
+		return nil
 	}
 	state, err := replay.StateForRevision(request.Range.Revision)
 	if err != nil {
-		t.Fatal(err)
+		if response.Error == model.ErrEtcdFutureRev.Error() {
+			return nil
+		}
+		lg.Error("Failed validating serializable operation", zap.Any("request", request), zap.Any("response", response))
+		return fmt.Errorf("request about a future rev with response")
 	}
 
 	_, expectResp := state.Step(request)
-	if !reflect.DeepEqual(response.EtcdResponse.Range, expectResp.Range) {
-		t.Errorf("Invalid serializable response, diff: %s", cmp.Diff(response.EtcdResponse.Range, expectResp.Range))
+
+	if diff := cmp.Diff(response.EtcdResponse.Range, expectResp.Range); diff != "" {
+		lg.Error("Failed validating serializable operation", zap.Any("request", request), zap.String("diff", diff))
+		return fmt.Errorf("response didn't match expected")
 	}
+	return nil
 }

tests/robustness/validate/operations_test.go

@@ -0,0 +1,295 @@
+// Copyright 2024 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//nolint:unparam
+package validate
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/anishathalye/porcupine"
+	"go.uber.org/zap/zaptest"
+
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+)
+
+func TestValidateSerializableOperations(t *testing.T) {
+	tcs := []struct {
+		name              string
+		persistedRequests []model.EtcdRequest
+		operations        []porcupine.Operation
+		expectError       string
+	}{
+		{
+			name: "Success",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input:  rangeRequest("a", "z", 1, 0),
+					Output: rangeResponse(0),
+				},
+				{
+					Input:  rangeRequest("a", "z", 2, 0),
+					Output: rangeResponse(1, keyValue("a", "1", 2)),
+				},
+				{
+					Input: rangeRequest("a", "z", 3, 0),
+					Output: rangeResponse(2,
+						keyValue("a", "1", 2),
+						keyValue("b", "2", 3),
+					),
+				},
+				{
+					Input: rangeRequest("a", "z", 4, 0),
+					Output: rangeResponse(3,
+						keyValue("a", "1", 2),
+						keyValue("b", "2", 3),
+						keyValue("c", "3", 4),
+					),
+				},
+				{
+					Input: rangeRequest("a", "z", 4, 3),
+					Output: rangeResponse(3,
+						keyValue("a", "1", 2),
+						keyValue("b", "2", 3),
+						keyValue("c", "3", 4),
+					),
+				},
+				{
+					Input: rangeRequest("a", "z", 4, 4),
+					Output: rangeResponse(3,
+						keyValue("a", "1", 2),
+						keyValue("b", "2", 3),
+						keyValue("c", "3", 4),
+					),
+				},
+				{
+					Input: rangeRequest("a", "z", 4, 2),
+					Output: rangeResponse(3,
+						keyValue("a", "1", 2),
+						keyValue("b", "2", 3),
+					),
+				},
+				{
+					Input: rangeRequest("b\x00", "z", 4, 2),
+					Output: rangeResponse(1,
+						keyValue("c", "3", 4),
+					),
+				},
+				{
+					Input: rangeRequest("b", "", 4, 0),
+					Output: rangeResponse(1,
+						keyValue("b", "2", 3),
+					),
+				},
+				{
+					Input:  rangeRequest("b", "", 2, 0),
+					Output: rangeResponse(0),
+				},
+			},
+		},
+		{
+			name: "Invalid order",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input: rangeRequest("a", "z", 4, 0),
+					Output: rangeResponse(3,
+						keyValue("c", "3", 4),
+						keyValue("b", "2", 3),
+						keyValue("a", "1", 2),
+					),
+				},
+			},
+			expectError: "response didn't match expected",
+		},
+		{
+			name: "Invalid count",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input:  rangeRequest("a", "z", 1, 0),
+					Output: rangeResponse(1),
+				},
+			},
+			expectError: "response didn't match expected",
+		},
+		{
+			name: "Invalid keys",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input: rangeRequest("a", "z", 2, 0),
+					Output: rangeResponse(3,
+						keyValue("b", "2", 3),
+					),
+				},
+			},
+			expectError: "response didn't match expected",
+		},
+		{
+			name: "Invalid revision",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input: rangeRequest("a", "z", 2, 0),
+					Output: rangeResponse(3,
+						keyValue("a", "1", 2),
+						keyValue("b", "2", 3),
+					),
+				},
+			},
+			expectError: "response didn't match expected",
+		},
+		{
+			name: "Error",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input:  rangeRequest("a", "z", 2, 0),
+					Output: errorResponse(model.ErrEtcdFutureRev),
+				},
+				{
+					Input:  rangeRequest("a", "z", 2, 0),
+					Output: errorResponse(fmt.Errorf("timeout")),
+				},
+			},
+		},
+		{
+			name: "Future rev returned",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input:  rangeRequest("a", "z", 6, 0),
+					Output: errorResponse(model.ErrEtcdFutureRev),
+				},
+			},
+		},
+		{
+			name: "Future rev success",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input:  rangeRequest("a", "z", 6, 0),
+					Output: rangeResponse(0),
+				},
+			},
+			expectError: "request about a future rev with response",
+		},
+		{
+			name: "Future rev failure",
+			persistedRequests: []model.EtcdRequest{
+				putRequest("a", "1"),
+				putRequest("b", "2"),
+				putRequest("c", "3"),
+			},
+			operations: []porcupine.Operation{
+				{
+					Input:  rangeRequest("a", "z", 6, 0),
+					Output: errorResponse(fmt.Errorf("timeout")),
+				},
+			},
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			replay := model.NewReplay(tc.persistedRequests)
+			err := validateSerializableOperations(zaptest.NewLogger(t), tc.operations, replay)
+			var errStr string
+			if err != nil {
+				errStr = err.Error()
+			}
+			if errStr != tc.expectError {
+				t.Errorf("validateSerializableOperations(...), got: %q, want: %q", err, tc.expectError)
+			}
+		})
+	}
+}
+
+func rangeRequest(start, end string, rev, limit int64) model.EtcdRequest {
+	return model.EtcdRequest{
+		Type: model.Range,
+		Range: &model.RangeRequest{
+			RangeOptions: model.RangeOptions{
+				Start: start,
+				End:   end,
+				Limit: limit,
+			},
+			Revision: rev,
+		},
+	}
+}
+
+func rangeResponse(count int64, kvs ...model.KeyValue) model.MaybeEtcdResponse {
+	if kvs == nil {
+		kvs = []model.KeyValue{}
+	}
+	return model.MaybeEtcdResponse{
+		EtcdResponse: model.EtcdResponse{
+			Range: &model.RangeResponse{
+				KVs:   kvs,
+				Count: count,
+			},
+		},
+	}
+}
+
+func errorResponse(err error) model.MaybeEtcdResponse {
+	return model.MaybeEtcdResponse{
+		Error: err.Error(),
+	}
+}
+
+func keyValue(key, value string, rev int64) model.KeyValue {
+	return model.KeyValue{
+		Key: key,
+		ValueRevision: model.ValueRevision{
+			Value:       model.ToValueOrHash(value),
+			ModRevision: rev,
+		},
+	}
+}

tests/robustness/validate/patch_history.go

@@ -24,7 +24,7 @@ import (
 	"go.etcd.io/etcd/tests/v3/robustness/traffic"
 )
 
-func patchedOperationHistory(reports []report.ClientReport, persistedRequests []model.EtcdRequest) []porcupine.Operation {
+func patchLinearizableOperations(reports []report.ClientReport, persistedRequests []model.EtcdRequest) []porcupine.Operation {
 	allOperations := relevantOperations(reports)
 	uniqueEvents := uniqueWatchEvents(reports)
 	operationsReturnTime := persistedOperationsReturnTime(allOperations, persistedRequests)

tests/robustness/validate/patch_history_test.go

@@ -388,7 +388,7 @@ func TestPatchHistory(t *testing.T) {
 			if tc.persistedRequest != nil {
 				requests = append(requests, *tc.persistedRequest)
 			}
-			operations := patchedOperationHistory([]report.ClientReport{
+			operations := patchLinearizableOperations([]report.ClientReport{
 				{
 					ClientID: 0,
 					KeyValue: history.History.Operations(),

tests/robustness/validate/validate.go

@@ -33,18 +33,25 @@ func ValidateAndReturnVisualize(t *testing.T, lg *zap.Logger, cfg Config, report
 	if err != nil {
 		t.Fatalf("Broken validation assumptions: %s", err)
 	}
-	patchedOperations := patchedOperationHistory(reports, persistedRequests)
-	linearizable, visualize := validateLinearizableOperationsAndVisualize(lg, patchedOperations, timeout)
+	linearizableOperations := patchLinearizableOperations(reports, persistedRequests)
+	serializableOperations := filterSerializableOperations(reports)
+
+	linearizable, visualize := validateLinearizableOperationsAndVisualize(lg, linearizableOperations, timeout)
 	if linearizable != porcupine.Ok {
 		t.Error("Failed linearization, skipping further validation")
 		return visualize
 	}
-	// TODO: Use requests from linearization instead of persisted requests from WAL.
-	err = validateWatch(lg, cfg, reports, persistedRequests)
+	// TODO: Use requests from linearization for replay.
+	replay := model.NewReplay(persistedRequests)
+
+	err = validateWatch(lg, cfg, reports, replay)
 	if err != nil {
 		t.Errorf("Failed validating watch history, err: %s", err)
 	}
-	validateSerializableOperations(t, lg, patchedOperations, persistedRequests)
+	err = validateSerializableOperations(lg, serializableOperations, replay)
+	if err != nil {
+		t.Errorf("Failed validating serializable operations, err: %s", err)
+	}
 	return visualize
 }
 

tests/robustness/validate/validate_test.go

@@ -1844,7 +1844,8 @@ func TestValidateWatch(t *testing.T) {
 	}
 	for _, tc := range tcs {
 		t.Run(tc.name, func(t *testing.T) {
-			err := validateWatch(zaptest.NewLogger(t), tc.config, tc.reports, tc.persistedRequests)
+			replay := model.NewReplay(tc.persistedRequests)
+			err := validateWatch(zaptest.NewLogger(t), tc.config, tc.reports, replay)
 			var errStr string
 			if err != nil {
 				errStr = err.Error()

tests/robustness/validate/watch.go

@@ -37,10 +37,9 @@ var (
 	errBrokeFilter       = errors.New("event not matching watch filter")
 )
 
-func validateWatch(lg *zap.Logger, cfg Config, reports []report.ClientReport, persistedRequests []model.EtcdRequest) error {
+func validateWatch(lg *zap.Logger, cfg Config, reports []report.ClientReport, replay *model.EtcdReplay) error {
 	lg.Info("Validating watch")
 	// Validate etcd watch properties defined in https://etcd.io/docs/v3.6/learning/api_guarantees/#watch-apis
-	replay := model.NewReplay(persistedRequests)
 	for _, r := range reports {
 		err := validateFilter(lg, r)
 		if err != nil {