refactor , logic improvement to check only forward updates are present, comments

Author: anandmukul93

agent/consul/acl_replication.go

@@ -66,10 +66,10 @@ type aclTypeReplicator interface {
 	// correction phase).
 	FetchUpdated(srv *Server, updates []string) (int, error)
 
-	//ensureUpdatedConsistent compares the updated items with the upsertable remotes
+	//ensureRemoteConsistent compares the updated items with the upsertable remotes
 	//returns consistent update check post stale batch read or other scenarios
 	//[]string of items missing from remote, []string of items updated remote, error if inconsistent
-	ensureUpdatedConsistent(updates []string) ([]string, []string, error)
+	ensureRemoteConsistent(updates []string) ([]string, []string, error)
 
 	// LenPendingUpdates should be the size of the data retrieved in
 	// FetchUpdated.
@@ -448,7 +448,10 @@ func (s *Server) replicateACLType(ctx context.Context, logger hclog.Logger, tr a
 			}
 			return 0, false, fmt.Errorf("failed to retrieve ACL %s updates: %v", tr.SingularNoun(), err)
 		}
-		_, _, err = tr.ensureUpdatedConsistent(res.LocalUpserts)
+		//if fetch updated gets stale inconsistent data then we should not proceed with applying
+		//the updates as that would lead to partial/stale data being replicated
+		//hence, we call ensureRemoteConsistent to validate the fetched updates with diff results
+		_, _, err = tr.ensureRemoteConsistent(res.LocalUpserts)
 		if err != nil {
 			if err == errContainsStaleData {
 				return 0, false, fmt.Errorf("failed to ensure consistent %s replication updates - stale data", tr.PluralNoun())

agent/consul/acl_replication_types.go

@@ -87,7 +87,7 @@ func (r *aclTokenReplicator) FetchUpdated(srv *Server, updates []string) (int, e
 	return len(r.updated), nil
 }
 
-func (r *aclTokenReplicator) ensureUpdatedConsistent(updates []string) ([]string, []string, error) {
+func (r *aclTokenReplicator) ensureRemoteConsistent(updates []string) ([]string, []string, error) {
 	//return true if consistent updates,
 	return []string{}, []string{}, nil
 }
@@ -192,7 +192,7 @@ func (r *aclPolicyReplicator) FetchUpdated(srv *Server, updates []string) (int,
 	return len(r.updated), nil
 }
 
-func (r *aclPolicyReplicator) ensureUpdatedConsistent(updates []string) ([]string, []string, error) {
+func (r *aclPolicyReplicator) ensureRemoteConsistent(updates []string) ([]string, []string, error) {
 	updatedMap := make(map[string]*structs.ACLPolicy)
 	for _, policy := range r.updated {
 		updatedMap[policy.ID] = policy
@@ -205,30 +205,30 @@ func (r *aclPolicyReplicator) ensureUpdatedConsistent(updates []string) ([]strin
 
 	//iterate over all updates array which are policy IDs check if the hash match for both
 	var consistent bool = true
-	var deleted []string
-	var updated []string
+	var remoteNotCreated []string
+	var remoteNotUpdated []string
 	var err error = nil
 
 	for _, policyID := range updates {
 		if updatedPolicy, ok := updatedMap[policyID]; ok {
 			if remotePolicy, ok := remoteMap[policyID]; ok {
 				if !bytes.Equal(updatedPolicy.Hash, remotePolicy.Hash) && updatedPolicy.ModifyIndex < remotePolicy.ModifyIndex {
 					// remote stale batch did not get modified policy than what local diff calculated
-					updated = append(updated, policyID)
+					remoteNotUpdated = append(remoteNotUpdated, policyID)
 					consistent = false
 				}
 			}
 		} else if remotePolicy, ok := remoteMap[policyID]; ok && remotePolicy.ModifyIndex == remotePolicy.CreateIndex {
 			// remote stale batch did not get the created policy than what local diff calculated
-			deleted = append(deleted, policyID)
+			remoteNotCreated = append(remoteNotCreated, policyID)
 			consistent = false
 		}
 	}
 
 	if !consistent {
 		err = errContainsStaleData
 	}
-	return deleted, updated, err
+	return remoteNotCreated, remoteNotUpdated, err
 }
 
 func (r *aclPolicyReplicator) DeleteLocalBatch(srv *Server, batch []string) error {
@@ -352,7 +352,7 @@ func (r *aclRoleReplicator) FetchUpdated(srv *Server, updates []string) (int, er
 	return len(r.updated), nil
 }
 
-func (r *aclRoleReplicator) ensureUpdatedConsistent(updates []string) ([]string, []string, error) {
+func (r *aclRoleReplicator) ensureRemoteConsistent(updates []string) ([]string, []string, error) {
 	//return true if consistent updates,
 	return []string{}, []string{}, nil
 }