chore(shard-distirbutor): extend info to debug assignment conflicts (#7506)

<!-- Describe what has changed in this PR -->
**What changed?**
Extending the transaction with operation in the else branch to get the
state in case the comparison fails

<!-- Tell your future self why have you made these changes -->
**Why?**
we need to understand why there is a conflict when we assign the shards
but it is not possible because etcd is not showing which is the failing
comparison, so we need to be able to understand where the conflict is
generated

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
unit tests that cover this case actually reported the mismatch

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

Signed-off-by: edigregorio <edigregorio@uber.com>

Author: eleonoradgr

service/sharddistributor/store/etcd/executorstore/etcdstore.go

@@ -348,7 +348,9 @@ func (s *executorStoreImpl) Subscribe(ctx context.Context, namespace string) (<-
 
 func (s *executorStoreImpl) AssignShards(ctx context.Context, namespace string, request store.AssignShardsRequest, guard store.GuardFunc) error {
 	var ops []clientv3.Op
+	var opsElse []clientv3.Op
 	var comparisons []clientv3.Cmp
+	comparisonMaps := make(map[string]int64)
 
 	statsUpdates, err := s.prepareShardStatisticsUpdates(ctx, namespace, request.NewState.ShardAssignments)
 	if err != nil {
@@ -363,10 +365,12 @@ func (s *executorStoreImpl) AssignShards(ctx context.Context, namespace string,
 		// Add a comparison to ensure the executor's assigned state hasn't changed
 		// This prevents deleting an executor that just received a shard assignment
 		comparisons = append(comparisons, clientv3.Compare(clientv3.ModRevision(executorStateKey), "=", expectedModRevision))
+		comparisonMaps[executorStateKey] = expectedModRevision
 
 		// Delete all keys for this executor
 		executorPrefix := etcdkeys.BuildExecutorIDPrefix(s.prefix, namespace, executorID)
 		ops = append(ops, clientv3.OpDelete(executorPrefix, clientv3.WithPrefix()))
+		opsElse = append(opsElse, clientv3.OpGet(executorStateKey))
 	}
 
 	// 2. Prepare operations to update executor states and shard ownership,
@@ -386,6 +390,8 @@ func (s *executorStoreImpl) AssignShards(ctx context.Context, namespace string,
 		ops = append(ops, clientv3.OpPut(executorStateKey, string(compressedValue)))
 
 		comparisons = append(comparisons, clientv3.Compare(clientv3.ModRevision(executorStateKey), "=", state.ModRevision))
+		comparisonMaps[executorStateKey] = state.ModRevision
+		opsElse = append(opsElse, clientv3.OpGet(executorStateKey))
 	}
 
 	if len(ops) == 0 {
@@ -405,10 +411,11 @@ func (s *executorStoreImpl) AssignShards(ctx context.Context, namespace string,
 
 	// 4. Create a nested transaction operation. This allows us to add our own 'If' (comparisons)
 	// and 'Then' (ops) logic that will only execute if the outer guard's 'If' condition passes.
+	// we catch what is the state in the else operations so we can identify which part of the condition failed
 	nestedTxnOp := clientv3.OpTxn(
 		comparisons, // Our IF conditions
 		ops,         // Our THEN operations
-		nil,         // Our ELSE operations
+		opsElse,     // Our ELSE operations
 	)
 
 	// 5. Add the nested transaction to the guarded transaction's THEN clause and commit.
@@ -429,10 +436,18 @@ func (s *executorStoreImpl) AssignShards(ctx context.Context, namespace string,
 	if len(txnResp.Responses) == 0 {
 		return fmt.Errorf("unexpected empty response from transaction")
 	}
+
 	nestedResp := txnResp.Responses[0].GetResponseTxn()
 	if !nestedResp.Succeeded {
 		// This means our revision checks failed.
-		return fmt.Errorf("%w: transaction failed, a shard may have been concurrently assigned", store.ErrVersionConflict)
+		failingRevisionString := ""
+		for _, keyValue := range nestedResp.Responses[0].GetResponseRange().Kvs {
+			expectedValue, ok := comparisonMaps[string(keyValue.Key)]
+			if !ok || expectedValue != keyValue.ModRevision {
+				failingRevisionString = failingRevisionString + fmt.Sprintf("{ key: %s, expected:%v, actual: %v }", string(keyValue.Key), expectedValue, keyValue.ModRevision)
+			}
+		}
+		return fmt.Errorf("%w: transaction failed, a shard may have been concurrently assigned, %v", store.ErrVersionConflict, failingRevisionString)
 	}
 
 	// Apply shard statistics updates outside the main transaction to stay within etcd's max operations per txn.