fix conflict fdbserver/ApplyMetadataMutation.cpp

fdbclient/CommitTransaction.h

@@ -187,17 +187,21 @@ struct CommitTransactionRef {
 	VectorRef<MutationRef> mutations;
 	Version read_snapshot;
 	bool report_conflicting_keys;
+	SpanID spanContext;
 
 	template <class Ar>
 	force_inline void serialize(Ar& ar) {
 		if constexpr (is_fb_function<Ar>) {
 			serializer(
-			    ar, read_conflict_ranges, write_conflict_ranges, mutations, read_snapshot, report_conflicting_keys);
+			    ar, read_conflict_ranges, write_conflict_ranges, mutations, read_snapshot, report_conflicting_keys, spanContext);
 		} else {
 			serializer(ar, read_conflict_ranges, write_conflict_ranges, mutations, read_snapshot);
 			if (ar.protocolVersion().hasReportConflictingKeys()) {
 				serializer(ar, report_conflicting_keys);
 			}
+			if (ar.protocolVersion().hasSpanContext()) {
+				serializer(ar, spanContext);
+			}
 		}
 	}
 

fdbserver/ApplyMetadataMutation.cpp

@@ -28,6 +28,7 @@
 #include "fdbserver/LogProtocolMessage.h"
 #include "fdbserver/LogSystem.h"
 #include "flow/Error.h"
+#include "flow/Trace.h"
 
 Reference<StorageInfo> getStorageInfo(UID id,
                                       std::map<UID, Reference<StorageInfo>>* storageCache,
@@ -231,6 +232,8 @@ private:
 			privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
 			TraceEvent("ServerTag", dbgid).detail("Server", id).detail("Tag", tag.toString());
 
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", "LogProtocolMessage");
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 			toCommit->addTag(tag);
 			toCommit->writeTypedMessage(LogProtocolMessage(), true);
 			toCommit->addTag(tag);
@@ -293,6 +296,7 @@ private:
 		// This is done to make the cache servers aware of the cached key-ranges
 		MutationRef privatized = m;
 		privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
+		TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 		toCommit->addTag(cacheTag);
 		toCommit->writeTypedMessage(privatized);
 	}
@@ -371,6 +375,7 @@ private:
 
 			Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssId)).get();
 			if (tagV.present()) {
+				TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 				toCommit->addTag(decodeServerTagValue(tagV.get()));
 				toCommit->writeTypedMessage(privatized);
 			}
@@ -400,6 +405,7 @@ private:
 		if (tagV.present()) {
 			MutationRef privatized = m;
 			privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 			toCommit->addTag(decodeServerTagValue(tagV.get()));
 			toCommit->writeTypedMessage(privatized);
 		}
@@ -518,10 +524,12 @@ private:
 		if (m.param1 == lastEpochEndKey) {
 			toCommit->addTags(allTags);
 			toCommit->writeTypedMessage(LogProtocolMessage(), true);
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", "LogProtocolMessage");
 		}
 
 		MutationRef privatized = m;
 		privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
+		TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", m.toString());
 		toCommit->addTags(allTags);
 		toCommit->writeTypedMessage(privatized);
 	}
@@ -648,6 +656,8 @@ private:
 					privatized.param1 = kv.key.withPrefix(systemKeys.begin, arena);
 					privatized.param2 = keyAfter(kv.key, arena).withPrefix(systemKeys.begin, arena);
 
+					TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
+
 					toCommit->addTag(decodeServerTagValue(kv.value));
 					toCommit->writeTypedMessage(privatized);
 				}
@@ -670,6 +680,7 @@ private:
 								privatized.param2 =
 								    keyAfter(maybeTssRange.begin, arena).withPrefix(systemKeys.begin, arena);
 
+								TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 								toCommit->addTag(decodeServerTagValue(tagV.get()));
 								toCommit->writeTypedMessage(privatized);
 							}
@@ -853,6 +864,7 @@ private:
 		if (Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssId)).get(); tagV.present()) {
 			MutationRef privatized = m;
 			privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 			toCommit->addTag(decodeServerTagValue(tagV.get()));
 			toCommit->writeTypedMessage(privatized);
 		}
@@ -878,6 +890,7 @@ private:
 
 					MutationRef privatized = m;
 					privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
+					TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", privatized.toString());
 					toCommit->addTag(decodeServerTagValue(tagV.get()));
 					toCommit->writeTypedMessage(privatized);
 				}
@@ -971,6 +984,8 @@ private:
 			}
 
 			// Add the tags to both begin and end mutations
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", mutationBegin.toString());
+			TraceEvent(SevDebug, "SendingPrivatized", dbgid).detail("M", mutationEnd.toString());
 			toCommit->addTags(allTags);
 			toCommit->writeTypedMessage(mutationBegin);
 			toCommit->addTags(allTags);

fdbserver/ApplyMetadataMutation.h

@@ -20,6 +20,7 @@
 
 #ifndef FDBSERVER_APPLYMETADATAMUTATION_H
 #define FDBSERVER_APPLYMETADATAMUTATION_H
+#include <cstddef>
 #pragma once
 
 #include "fdbclient/BackupAgent.actor.h"
@@ -45,6 +46,7 @@ struct ResolverData {
 	LogPushData* toCommit = nullptr;
 	Version popVersion = 0; // exclusive, usually set to commitVersion + 1
 	std::map<UID, Reference<StorageInfo>>* storageCache = nullptr;
+	std::unordered_map<UID, StorageServerInterface>* tssMapping = nullptr;
 
 	// For initial broadcast
 	ResolverData(UID debugId, IKeyValueStore* store, KeyRangeMap<ServerCacheInfo>* info)
@@ -57,9 +59,10 @@ struct ResolverData {
 	             KeyRangeMap<ServerCacheInfo>* info,
 	             LogPushData* toCommit,
 	             Version popVersion,
-	             std::map<UID, Reference<StorageInfo>>* storageCache)
+	             std::map<UID, Reference<StorageInfo>>* storageCache,
+	             std::unordered_map<UID, StorageServerInterface>* tssMapping)
 	  : dbgid(debugId), txnStateStore(store), keyInfo(info), logSystem(logSystem), toCommit(toCommit),
-	    popVersion(popVersion), storageCache(storageCache) {}
+	    popVersion(popVersion), storageCache(storageCache), tssMapping(tssMapping) {}
 };
 
 inline bool isMetadataMutation(MutationRef const& m) {

fdbserver/CommitProxyServer.actor.cpp

@@ -180,12 +180,16 @@ struct ResolutionRequestBuilder {
 				getOutTransaction(resolver, trIn.read_snapshot)
 				    .write_conflict_ranges.push_back(requests[resolver].arena, r);
 		}
-		if (isTXNStateTransaction)
+		if (isTXNStateTransaction) {
 			for (int r = 0; r < requests.size(); r++) {
 				int transactionNumberInRequest =
 				    &getOutTransaction(r, trIn.read_snapshot) - requests[r].transactions.begin();
 				requests[r].txnStateTransactions.push_back(requests[r].arena, transactionNumberInRequest);
 			}
+			// Note only Resolver 0 got the correct spanContext, which means
+			// the reply from Resolver 0 has the right one back.
+			getOutTransaction(0, trIn.read_snapshot).spanContext = trRequest.spanContext;
+		}
 
 		std::vector<int> resolversUsed;
 		for (int r = 0; r < outTr.size(); r++)
@@ -941,7 +945,7 @@ ACTOR Future<Void> applyMetadataToCommittedTransactions(CommitBatchContext* self
 	ResolveTransactionBatchReply& reply = self->resolution[0];
 	ASSERT_WE_THINK(privateMutations.size() == reply.privateMutations.size());
 	for (int i = 0; i < privateMutations.size(); i++) {
-		std::cout << i << "\n" << printable(privateMutations[i]) << "\n" << printable(reply.privateMutations[i]) << "\n\n";
+		// std::cout << i << "\n" << printable(privateMutations[i]) << "\n" << printable(reply.privateMutations[i]) << "\n\n";
 		ASSERT_WE_THINK(privateMutations[i] == reply.privateMutations[i]);
 	}
 	if (self->forceRecovery) {

fdbserver/Resolver.actor.cpp

@@ -72,6 +72,7 @@ struct Resolver : ReferenceCounted<Resolver> {
 
 	std::map<UID, Reference<StorageInfo>> storageCache;
 	KeyRangeMap<ServerCacheInfo> keyInfo; // keyrange -> all storage servers in all DCs for the keyrange
+	std::unordered_map<UID, StorageServerInterface> tssMapping;
 
 	Version debugMinRecentStateVersion;
 
@@ -243,7 +244,8 @@ ACTOR Future<Void> resolveBatch(Reference<Resolver> self, ResolveTransactionBatc
 		                          &self->keyInfo,
 		                          &toCommit,
 		                          req.version + 1,
-		                          &self->storageCache);
+		                          &self->storageCache,
+		                          &self->tssMapping);
 		for (int t : req.txnStateTransactions) {
 			stateMutations += req.transactions[t].mutations.size();
 			stateBytes += req.transactions[t].mutations.expectedSize();
@@ -254,7 +256,7 @@ ACTOR Future<Void> resolveBatch(Reference<Resolver> self, ResolveTransactionBatc
 
 			// Generate private mutations for metadata mutations
 			if (reply.committed[t] == ConflictBatch::TransactionCommitted) {
-				applyMetadataMutations(SpanID(), resolverData, req.transactions[t].mutations);
+				applyMetadataMutations(req.transactions[t].spanContext, resolverData, req.transactions[t].mutations);
 			}
 		}