tests/robustness: Extract validation to separate package

tests/robustness/linearizability_test.go

@@ -26,8 +26,8 @@ import (
 	"go.etcd.io/etcd/api/v3/version"
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
-	"go.etcd.io/etcd/tests/v3/robustness/model"
 	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+	"go.etcd.io/etcd/tests/v3/robustness/validate"
 )
 
 func TestRobustness(t *testing.T) {
@@ -65,10 +65,7 @@ func TestRobustness(t *testing.T) {
 			name:      traffic.Name + "ClusterOfSize3",
 			failpoint: RandomFailpoint,
 			traffic:   traffic,
-			watch: watchConfig{
-				expectUniqueRevision: traffic.Traffic.ExpectUniqueRevision(),
-			},
-			cluster: *e2e.NewConfig(clusterOfSize3Options...),
+			cluster:   *e2e.NewConfig(clusterOfSize3Options...),
 		})
 	}
 	scenarios = append(scenarios, testScenario{
@@ -160,7 +157,8 @@ func testRobustness(ctx context.Context, t *testing.T, lg *zap.Logger, s testSce
 
 	watchProgressNotifyEnabled := r.clus.Cfg.WatchProcessNotifyInterval != 0
 	validateGotAtLeastOneProgressNotify(t, r.clientReports, s.watch.requestProgress || watchProgressNotifyEnabled)
-	r.visualizeHistory = validateCorrectness(t, lg, s.watch, r.clientReports)
+	validateConfig := validate.Config{ExpectRevisionUnique: s.traffic.Traffic.ExpectUniqueRevision()}
+	r.visualizeHistory = validate.ValidateAndReturnVisualize(t, lg, validateConfig, r.clientReports)
 
 	panicked = false
 }
@@ -213,9 +211,3 @@ func forcestopCluster(clus *e2e.EtcdProcessCluster) error {
 	}
 	return clus.ConcurrentStop()
 }
-
-func validateCorrectness(t *testing.T, lg *zap.Logger, cfg watchConfig, reports []traffic.ClientReport) (visualize func(basepath string)) {
-	validateWatchCorrectness(t, cfg, reports)
-	operations := operationsFromClientReports(reports)
-	return model.ValidateOperationHistoryAndReturnVisualize(t, lg, operations)
-}

tests/robustness/model/history.go

@@ -16,36 +16,16 @@ package model
 
 import (
 	"fmt"
-	"testing"
 	"time"
 
 	"github.com/anishathalye/porcupine"
-	"go.uber.org/zap"
 
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
 )
 
-// ValidateOperationHistoryAndReturnVisualize return visualize as porcupine.linearizationInfo used to generate visualization is private.
-func ValidateOperationHistoryAndReturnVisualize(t *testing.T, lg *zap.Logger, operations []porcupine.Operation) (visualize func(basepath string)) {
-	linearizable, info := porcupine.CheckOperationsVerbose(NonDeterministicModel, operations, 5*time.Minute)
-	if linearizable == porcupine.Illegal {
-		t.Error("Model is not linearizable")
-	}
-	if linearizable == porcupine.Unknown {
-		t.Error("Linearization timed out")
-	}
-	return func(path string) {
-		lg.Info("Saving visualization", zap.String("path", path))
-		err := porcupine.VisualizePath(NonDeterministicModel, info, path)
-		if err != nil {
-			t.Errorf("Failed to visualize, err: %v", err)
-		}
-	}
-}
-
 // AppendableHistory allows to collect history of sequential operations.
 //
 // Ensures that operation history is compatible with porcupine library, by preventing concurrent requests sharing the

tests/robustness/validate/operations.go

@@ -0,0 +1,42 @@
+// Copyright 2023 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.
+
+package validate
+
+import (
+	"testing"
+	"time"
+
+	"github.com/anishathalye/porcupine"
+	"go.uber.org/zap"
+
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+)
+
+func validateOperationHistoryAndReturnVisualize(t *testing.T, lg *zap.Logger, operations []porcupine.Operation) (visualize func(basepath string)) {
+	linearizable, info := porcupine.CheckOperationsVerbose(model.NonDeterministicModel, operations, 5*time.Minute)
+	if linearizable == porcupine.Illegal {
+		t.Error("Model is not linearizable")
+	}
+	if linearizable == porcupine.Unknown {
+		t.Error("Linearization timed out")
+	}
+	return func(path string) {
+		lg.Info("Saving visualization", zap.String("path", path))
+		err := porcupine.VisualizePath(model.NonDeterministicModel, info, path)
+		if err != nil {
+			t.Errorf("Failed to visualize, err: %v", err)
+		}
+	}
+}

tests/robustness/validate/patch_history.go

@@ -0,0 +1,107 @@
+// Copyright 2023 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.
+
+package validate
+
+import (
+	"github.com/anishathalye/porcupine"
+
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+)
+
+func patchOperationsWithWatchEvents(operations []porcupine.Operation, watchEvents map[model.EtcdOperation]traffic.TimedWatchEvent) []porcupine.Operation {
+	newOperations := make([]porcupine.Operation, 0, len(operations))
+	lastObservedOperation := lastOperationObservedInWatch(operations, watchEvents)
+
+	for _, op := range operations {
+		request := op.Input.(model.EtcdRequest)
+		resp := op.Output.(model.EtcdNonDeterministicResponse)
+		if resp.Err == nil || op.Call > lastObservedOperation.Call || request.Type != model.Txn {
+			// Cannot patch those requests.
+			newOperations = append(newOperations, op)
+			continue
+		}
+		event := matchWatchEvent(request.Txn, watchEvents)
+		if event != nil {
+			// Set revision and time based on watchEvent.
+			op.Return = event.Time.Nanoseconds()
+			op.Output = model.EtcdNonDeterministicResponse{
+				EtcdResponse:  model.EtcdResponse{Revision: event.Revision},
+				ResultUnknown: true,
+			}
+			newOperations = append(newOperations, op)
+			continue
+		}
+		if hasNonUniqueWriteOperation(request.Txn) && !hasUniqueWriteOperation(request.Txn) {
+			// Leave operation as it is as we cannot match non-unique operations to watch events.
+			newOperations = append(newOperations, op)
+			continue
+		}
+		// Remove non persisted operations
+	}
+	return newOperations
+}
+
+func lastOperationObservedInWatch(operations []porcupine.Operation, watchEvents map[model.EtcdOperation]traffic.TimedWatchEvent) porcupine.Operation {
+	var maxCallTime int64
+	var lastOperation porcupine.Operation
+	for _, op := range operations {
+		request := op.Input.(model.EtcdRequest)
+		if request.Type != model.Txn {
+			continue
+		}
+		event := matchWatchEvent(request.Txn, watchEvents)
+		if event != nil && op.Call > maxCallTime {
+			maxCallTime = op.Call
+			lastOperation = op
+		}
+	}
+	return lastOperation
+}
+
+func matchWatchEvent(request *model.TxnRequest, watchEvents map[model.EtcdOperation]traffic.TimedWatchEvent) *traffic.TimedWatchEvent {
+	for _, etcdOp := range append(request.OperationsOnSuccess, request.OperationsOnFailure...) {
+		if etcdOp.Type == model.Put {
+			// Remove LeaseID which is not exposed in watch.
+			event, ok := watchEvents[model.EtcdOperation{
+				Type:  etcdOp.Type,
+				Key:   etcdOp.Key,
+				Value: etcdOp.Value,
+			}]
+			if ok {
+				return &event
+			}
+		}
+	}
+	return nil
+}
+
+func hasNonUniqueWriteOperation(request *model.TxnRequest) bool {
+	for _, etcdOp := range request.OperationsOnSuccess {
+		if etcdOp.Type == model.Put || etcdOp.Type == model.Delete {
+			return true
+		}
+	}
+	return false
+}
+
+func hasUniqueWriteOperation(request *model.TxnRequest) bool {
+	for _, etcdOp := range request.OperationsOnSuccess {
+		if etcdOp.Type == model.Put {
+			return true
+		}
+	}
+	return false
+}

tests/robustness/validate/validate.go

@@ -0,0 +1,58 @@
+// Copyright 2023 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.
+
+package validate
+
+import (
+	"testing"
+
+	"github.com/anishathalye/porcupine"
+	"go.uber.org/zap"
+
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+)
+
+// ValidateAndReturnVisualize return visualize as porcupine.linearizationInfo used to generate visualization is private.
+func ValidateAndReturnVisualize(t *testing.T, lg *zap.Logger, cfg Config, reports []traffic.ClientReport) (visualize func(basepath string)) {
+	validateWatch(t, cfg, reports)
+	allOperations := operations(reports)
+	watchEvents := uniqueWatchEvents(reports)
+	newOperations := patchOperationsWithWatchEvents(allOperations, watchEvents)
+	return validateOperationHistoryAndReturnVisualize(t, lg, newOperations)
+}
+
+func operations(reports []traffic.ClientReport) []porcupine.Operation {
+	var ops []porcupine.Operation
+	for _, r := range reports {
+		ops = append(ops, r.OperationHistory.Operations()...)
+	}
+	return ops
+}
+
+func uniqueWatchEvents(reports []traffic.ClientReport) map[model.EtcdOperation]traffic.TimedWatchEvent {
+	persisted := map[model.EtcdOperation]traffic.TimedWatchEvent{}
+	for _, r := range reports {
+		for _, resp := range r.Watch {
+			for _, event := range resp.Events {
+				persisted[event.Op] = traffic.TimedWatchEvent{Time: resp.Time, WatchEvent: event}
+			}
+		}
+	}
+	return persisted
+}
+
+type Config struct {
+	ExpectRevisionUnique bool
+}