From dfd5c3e195a744ae7c36768b98d1cba54d500732 Mon Sep 17 00:00:00 2001 From: Marek Siarkowicz Date: Sat, 22 Jun 2024 14:08:02 +0200 Subject: [PATCH] Separate persisted responses without knowing their revision to prevent duplicating state during linearization Signed-off-by: Marek Siarkowicz --- tests/robustness/model/describe.go | 7 +- tests/robustness/model/deterministic.go | 28 +++-- tests/robustness/model/deterministic_test.go | 102 +++++++++++++++++++ tests/robustness/model/history.go | 2 +- tests/robustness/model/non_deterministic.go | 32 ++++-- tests/robustness/validate/operations.go | 2 +- tests/robustness/validate/patch_history.go | 27 +++-- 7 files changed, 168 insertions(+), 32 deletions(-) diff --git a/tests/robustness/model/describe.go b/tests/robustness/model/describe.go index a889b4d2b1b9..a8c3bc1556d7 100644 --- a/tests/robustness/model/describe.go +++ b/tests/robustness/model/describe.go @@ -28,8 +28,11 @@ func describeEtcdResponse(request EtcdRequest, response MaybeEtcdResponse) strin if response.ClientError != "" { return fmt.Sprintf("err: %q", response.ClientError) } - if response.PartialResponse { - return fmt.Sprintf("unknown, rev: %d", response.Revision) + if response.Persisted { + if response.PersistedRevision != 0 { + return fmt.Sprintf("unknown, rev: %d", response.PersistedRevision) + } + return fmt.Sprintf("unknown") } switch request.Type { case Range: diff --git a/tests/robustness/model/deterministic.go b/tests/robustness/model/deterministic.go index 49111f66835f..7a17da118848 100644 --- a/tests/robustness/model/deterministic.go +++ b/tests/robustness/model/deterministic.go @@ -123,7 +123,7 @@ func (s EtcdState) Step(request EtcdRequest) (EtcdState, MaybeEtcdResponse) { if request.Range.Revision < newState.CompactRevision { return newState, MaybeEtcdResponse{EtcdResponse: EtcdResponse{ClientError: mvcc.ErrCompacted.Error()}} } - return newState, MaybeEtcdResponse{PartialResponse: true, EtcdResponse: EtcdResponse{Revision: newState.Revision}} + return newState, MaybeEtcdResponse{Persisted: true, PersistedRevision: newState.Revision} case Txn: failure := false for _, cond := range request.Txn.Conditions { @@ -351,15 +351,17 @@ type LeaseRevokeRequest struct { } type DefragmentRequest struct{} -// MaybeEtcdResponse extends EtcdResponse to represent partial or failed responses. -// Possible states: -// * Normal response. Only EtcdResponse is set. -// * Partial response. The EtcdResponse.Revision and PartialResponse are set. -// * Failed response. Only Err is set. +// MaybeEtcdResponse extends EtcdResponse to include partial information about responses to a request. +// Possible response state information: +// * Normal response. Client observed response. Only EtcdResponse is set. +// * Persisted. Client didn't observe response, but we know it was persisted by etcd. Only Persisted is set +// * Persisted with Revision. Client didn't observe response, but we know that it was persisted, and it's revision. Both Persisted and PersistedRevision is set. +// * Error response. Client observed error, but we don't know if it was persisted. Only Error is set. type MaybeEtcdResponse struct { EtcdResponse - PartialResponse bool - Error string + Persisted bool + PersistedRevision int64 + Error string } var ErrEtcdFutureRev = errors.New("future rev") @@ -376,7 +378,15 @@ type EtcdResponse struct { } func Match(r1, r2 MaybeEtcdResponse) bool { - return ((r1.PartialResponse || r2.PartialResponse) && (r1.Revision == r2.Revision)) || reflect.DeepEqual(r1, r2) + r1Revision := r1.Revision + if r1.Persisted { + r1Revision = r1.PersistedRevision + } + r2Revision := r2.Revision + if r2.Persisted { + r2Revision = r2.PersistedRevision + } + return (r1.Persisted && r1.PersistedRevision == 0) || (r2.Persisted && r2.PersistedRevision == 0) || ((r1.Persisted || r2.Persisted) && (r1.Error != "" || r2.Error != "" || r1Revision == r2Revision)) || reflect.DeepEqual(r1, r2) } type TxnResponse struct { diff --git a/tests/robustness/model/deterministic_test.go b/tests/robustness/model/deterministic_test.go index 965eb2ecd056..4f2bfa8d3b1f 100644 --- a/tests/robustness/model/deterministic_test.go +++ b/tests/robustness/model/deterministic_test.go @@ -19,6 +19,7 @@ import ( "testing" "github.com/google/go-cmp/cmp" + "github.com/stretchr/testify/assert" "go.etcd.io/etcd/api/v3/mvccpb" ) @@ -406,3 +407,104 @@ var commonTestScenarios = []modelTestCase{ }, }, } + +func TestMaybeEtcdResponseMatch(t *testing.T) { + tcs := []struct { + r1 MaybeEtcdResponse + r2 MaybeEtcdResponse + expectMatch bool + }{ + { + r1: MaybeEtcdResponse{}, + r2: MaybeEtcdResponse{}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1}}, + r2: MaybeEtcdResponse{}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false}}}, + r2: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1}}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false}}}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 0}}}}}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{Error: "error"}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{Error: "error1"}, + r2: MaybeEtcdResponse{Error: "error2"}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{Error: "error"}, + r2: MaybeEtcdResponse{Error: "error"}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{Persisted: true}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{Persisted: true, PersistedRevision: 1}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: 1, Txn: &TxnResponse{Failure: false, Results: []EtcdOperationResult{{Deleted: 1}}}}}, + r2: MaybeEtcdResponse{Persisted: true, PersistedRevision: 2}, + expectMatch: false, + }, + { + r1: MaybeEtcdResponse{Error: "error"}, + r2: MaybeEtcdResponse{Persisted: true}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{Error: "error"}, + r2: MaybeEtcdResponse{Persisted: true, PersistedRevision: 2}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{Persisted: true}, + r2: MaybeEtcdResponse{Persisted: true, PersistedRevision: 2}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{Persisted: true, PersistedRevision: 2}, + r2: MaybeEtcdResponse{Persisted: true, PersistedRevision: 2}, + expectMatch: true, + }, + { + r1: MaybeEtcdResponse{Persisted: true, PersistedRevision: 1}, + r2: MaybeEtcdResponse{Persisted: true, PersistedRevision: 2}, + expectMatch: false, + }, + } + for _, tc := range tcs { + gotMatch := Match(tc.r1, tc.r2) + assert.Equal(t, tc.expectMatch, gotMatch, "Reported invalid match for %+v and %+v", tc.r1, tc.r2) + gotReverseMatch := Match(tc.r2, tc.r1) + assert.Equal(t, gotMatch, gotReverseMatch, "Match(r1,r2) != Match(r2, r1)") + } + +} diff --git a/tests/robustness/model/history.go b/tests/robustness/model/history.go index c64e1a73aa9c..fd1b051294c5 100644 --- a/tests/robustness/model/history.go +++ b/tests/robustness/model/history.go @@ -365,7 +365,7 @@ func failedResponse(err error) MaybeEtcdResponse { } func partialResponse(revision int64) MaybeEtcdResponse { - return MaybeEtcdResponse{PartialResponse: true, EtcdResponse: EtcdResponse{Revision: revision}} + return MaybeEtcdResponse{Persisted: true, PersistedRevision: revision} } func putRequest(key, value string) EtcdRequest { diff --git a/tests/robustness/model/non_deterministic.go b/tests/robustness/model/non_deterministic.go index f6503324a9d1..3167e340fd16 100644 --- a/tests/robustness/model/non_deterministic.go +++ b/tests/robustness/model/non_deterministic.go @@ -58,17 +58,19 @@ func (states nonDeterministicState) apply(request EtcdRequest, response MaybeEtc var newStates nonDeterministicState switch { case response.Error != "": - newStates = states.stepFailedResponse(request) - case response.PartialResponse: - newStates = states.applyResponseRevision(request, response.EtcdResponse.Revision) + newStates = states.applyFailedRequest(request) + case response.Persisted && response.PersistedRevision == 0: + newStates = states.applyPersistedRequest(request) + case response.Persisted && response.PersistedRevision != 0: + newStates = states.applyPersistedRequestWithRevision(request, response.PersistedRevision) default: - newStates = states.applySuccessfulResponse(request, response.EtcdResponse) + newStates = states.applyRequestWithResponse(request, response.EtcdResponse) } return len(newStates) > 0, newStates } -// stepFailedResponse duplicates number of states by considering both cases, request was persisted and request was lost. -func (states nonDeterministicState) stepFailedResponse(request EtcdRequest) nonDeterministicState { +// applyFailedRequest returns both the original states and states with applied request. It considers both cases, request was persisted and request was lost. +func (states nonDeterministicState) applyFailedRequest(request EtcdRequest) nonDeterministicState { newStates := make(nonDeterministicState, 0, len(states)*2) for _, s := range states { newStates = append(newStates, s) @@ -80,8 +82,18 @@ func (states nonDeterministicState) stepFailedResponse(request EtcdRequest) nonD return newStates } -// applyResponseRevision filters possible states by leaving ony states that would return proper revision. -func (states nonDeterministicState) applyResponseRevision(request EtcdRequest, responseRevision int64) nonDeterministicState { +// applyPersistedRequest applies request to all possible states. +func (states nonDeterministicState) applyPersistedRequest(request EtcdRequest) nonDeterministicState { + newStates := make(nonDeterministicState, 0, len(states)) + for _, s := range states { + newState, _ := s.Step(request) + newStates = append(newStates, newState) + } + return newStates +} + +// applyPersistedRequestWithRevision applies request to all possible states, but leaves only states that would return proper revision. +func (states nonDeterministicState) applyPersistedRequestWithRevision(request EtcdRequest, responseRevision int64) nonDeterministicState { newStates := make(nonDeterministicState, 0, len(states)) for _, s := range states { newState, modelResponse := s.Step(request) @@ -92,8 +104,8 @@ func (states nonDeterministicState) applyResponseRevision(request EtcdRequest, r return newStates } -// applySuccessfulResponse filters possible states by leaving ony states that would respond correctly. -func (states nonDeterministicState) applySuccessfulResponse(request EtcdRequest, response EtcdResponse) nonDeterministicState { +// applyRequestWithResponse applies request to all possible states, but leaves only state that would return proper response. +func (states nonDeterministicState) applyRequestWithResponse(request EtcdRequest, response EtcdResponse) nonDeterministicState { newStates := make(nonDeterministicState, 0, len(states)) for _, s := range states { newState, modelResponse := s.Step(request) diff --git a/tests/robustness/validate/operations.go b/tests/robustness/validate/operations.go index a2cc10d4134f..efb364080447 100644 --- a/tests/robustness/validate/operations.go +++ b/tests/robustness/validate/operations.go @@ -83,7 +83,7 @@ func filterSerializableOperations(clients []report.ClientReport) []porcupine.Ope } func validateSerializableRead(lg *zap.Logger, replay *model.EtcdReplay, request model.EtcdRequest, response model.MaybeEtcdResponse) error { - if response.PartialResponse || response.Error != "" { + if response.Persisted || response.Error != "" { return nil } state, err := replay.StateForRevision(request.Range.Revision) diff --git a/tests/robustness/validate/patch_history.go b/tests/robustness/validate/patch_history.go index c5e1efa2b613..5fe3271eac9b 100644 --- a/tests/robustness/validate/patch_history.go +++ b/tests/robustness/validate/patch_history.go @@ -71,11 +71,12 @@ func patchOperations(operations []porcupine.Operation, watchEvents map[model.Eve for _, op := range operations { request := op.Input.(model.EtcdRequest) resp := op.Output.(model.MaybeEtcdResponse) - if resp.Error == "" || request.Type != model.Txn { + if resp.Error == "" || request.Type != model.Txn || !hasWriteTxn(request.Txn) { // Cannot patch those requests. newOperations = append(newOperations, op) continue } + var resourceVersion int64 = 0 if op.Call <= lastObservedOperation.Call { matchingEvent := matchWatchEvent(request.Txn, watchEvents) if matchingEvent != nil { @@ -84,7 +85,7 @@ func patchOperations(operations []porcupine.Operation, watchEvents map[model.Eve if eventTime < op.Return { op.Return = eventTime } - op.Output = model.MaybeEtcdResponse{PartialResponse: true, EtcdResponse: model.EtcdResponse{Revision: matchingEvent.Revision}} + resourceVersion = matchingEvent.Revision } } persistedReturnTime := matchReturnTime(request, persistedOperations) @@ -94,9 +95,17 @@ func patchOperations(operations []porcupine.Operation, watchEvents map[model.Eve op.Return = *persistedReturnTime } } - if persistedReturnTime == nil && canBeDiscarded(request.Txn) { - // Remove non persisted operations - continue + if isUniqueTxn(request.Txn) { + if persistedReturnTime == nil { + // Remove non persisted operations + continue + } else { + if resourceVersion != 0 { + op.Output = model.MaybeEtcdResponse{Persisted: true, PersistedRevision: resourceVersion} + } else { + op.Output = model.MaybeEtcdResponse{Persisted: true} + } + } } // Leave operation as it is as we cannot discard it. newOperations = append(newOperations, op) @@ -137,12 +146,12 @@ func matchWatchEvent(request *model.TxnRequest, watchEvents map[model.Event]clie return nil } -func canBeDiscarded(request *model.TxnRequest) bool { - return operationsCanBeDiscarded(request.OperationsOnSuccess) && operationsCanBeDiscarded(request.OperationsOnFailure) +func isUniqueTxn(request *model.TxnRequest) bool { + return hasUniqueWriteOperation(request.OperationsOnSuccess) || hasUniqueWriteOperation(request.OperationsOnFailure) } -func operationsCanBeDiscarded(ops []model.EtcdOperation) bool { - return hasUniqueWriteOperation(ops) || !hasWriteOperation(ops) +func hasWriteTxn(txn *model.TxnRequest) bool { + return hasWriteOperation(txn.OperationsOnSuccess) || hasWriteOperation(txn.OperationsOnFailure) } func hasWriteOperation(ops []model.EtcdOperation) bool {