fix: reply promise streams could receive messages out of order if the sender and receiver were the same process

fdbrpc/FlowTransport.actor.cpp

@@ -919,11 +919,11 @@ ACTOR static void deliver(TransportData* self,
                           TaskPriority priority,
                           ArenaReader reader,
                           bool inReadSocket) {
-	// We want to run the task at the right priority. If the priority
-	// is higher than the current priority (which is ReadSocket) we
-	// can just upgrade. Otherwise we'll context switch so that we
-	// don't block other tasks that might run with a higher priority.
-	if (priority < TaskPriority::ReadSocket || !inReadSocket) {
+	// We want to run the task at the right priority. If the priority is higher than the current priority (which is
+	// ReadSocket) we can just upgrade. Otherwise we'll context switch so that we don't block other tasks that might run
+	// with a higher priority. ReplyPromiseStream needs to guarentee that messages are recieved in the order they were
+	// sent, so even in the case of local delivery those messages need to skip this delay.
+	if (priority < TaskPriority::ReadSocket || (priority != TaskPriority::NoDeliverDelay && !inReadSocket)) {
 		wait(delay(0, priority));
 	} else {
 		g_network->setCurrentTask(priority);

fdbrpc/fdbrpc.h

@@ -361,7 +361,7 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
 					FlowTransport::transport().sendUnreliable(
 					    SerializeSource<ErrorOr<AcknowledgementReply>>(
 					        AcknowledgementReply(acknowledgements.bytesAcknowledged)),
-					    acknowledgements.getEndpoint(TaskPriority::ReadSocket),
+					    acknowledgements.getEndpoint(TaskPriority::NoDeliverDelay),
 					    false);
 				}
 			}
@@ -378,7 +378,7 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
 			acknowledgements.bytesAcknowledged += res.expectedSize();
 			FlowTransport::transport().sendUnreliable(SerializeSource<ErrorOr<AcknowledgementReply>>(
 			                                              AcknowledgementReply(acknowledgements.bytesAcknowledged)),
-			                                          acknowledgements.getEndpoint(TaskPriority::ReadSocket),
+			                                          acknowledgements.getEndpoint(TaskPriority::NoDeliverDelay),
 			                                          false);
 		}
 		return res;
@@ -389,13 +389,13 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
 			// Notify the server that a client is not using this ReplyPromiseStream anymore
 			FlowTransport::transport().sendUnreliable(
 			    SerializeSource<ErrorOr<AcknowledgementReply>>(operation_obsolete()),
-			    acknowledgements.getEndpoint(TaskPriority::ReadSocket),
+			    acknowledgements.getEndpoint(TaskPriority::NoDeliverDelay),
 			    false);
 		}
 		if (isRemoteEndpoint() && !sentError && !acknowledgements.failures.isReady()) {
 			// The ReplyPromiseStream was cancelled before sending an error, so the storage server must have died
 			FlowTransport::transport().sendUnreliable(SerializeSource<ErrorOr<EnsureTable<T>>>(broken_promise()),
-			                                          getEndpoint(TaskPriority::ReadSocket),
+			                                          getEndpoint(TaskPriority::NoDeliverDelay),
 			                                          false);
 		}
 	}
@@ -406,7 +406,7 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
 template <class T>
 class ReplyPromiseStream {
 public:
-	// The endpoints of a ReplyPromiseStream must be initialized at Task::ReadSocket, because with lower priorities a
+	// The endpoints of a ReplyPromiseStream must be initialized at Task::NoDeliverDelay, because a
 	// delay(0) in FlowTransport deliver can cause out of order delivery.
 
 	// stream.send( request )
@@ -416,7 +416,7 @@ public:
 	void send(U&& value) const {
 		if (queue->isRemoteEndpoint()) {
 			if (!queue->acknowledgements.getRawEndpoint().isValid()) {
-				value.acknowledgeToken = queue->acknowledgements.getEndpoint(TaskPriority::ReadSocket).token;
+				value.acknowledgeToken = queue->acknowledgements.getEndpoint(TaskPriority::NoDeliverDelay).token;
 			}
 			queue->acknowledgements.bytesSent += value.expectedSize();
 			FlowTransport::transport().sendUnreliable(
@@ -477,7 +477,7 @@ public:
 			errors->delPromiseRef();
 	}
 
-	const Endpoint& getEndpoint() const { return queue->getEndpoint(TaskPriority::ReadSocket); }
+	const Endpoint& getEndpoint() const { return queue->getEndpoint(TaskPriority::NoDeliverDelay); }
 
 	bool operator==(const ReplyPromiseStream<T>& rhs) const { return queue == rhs.queue; }
 	bool isEmpty() const { return !queue->isReady(); }

fdbserver/storageserver.actor.cpp

@@ -3891,7 +3891,7 @@ ACTOR Future<Void> update(StorageServer* data, bool* pReceivedUpdate) {
 
 		if (ver != invalidVersion && ver > data->version.get()) {
 			// TODO(alexmiller): Update to version tracking.
-			DEBUG_KEY_RANGE("SSUpdate", ver, KeyRangeRef());
+			// DEBUG_KEY_RANGE("SSUpdate", ver, KeyRangeRef());
 
 			data->mutableData().createNewVersion(ver);
 			if (data->otherError.getFuture().isReady())
@@ -4179,7 +4179,7 @@ bool StorageServerDisk::makeVersionMutationsDurable(Version& prevStorageVersion,
 		VerUpdateRef const& v = u->second;
 		ASSERT(v.version > prevStorageVersion && v.version <= newStorageVersion);
 		// TODO(alexmiller): Update to version tracking.
-		DEBUG_KEY_RANGE("makeVersionMutationsDurable", v.version, KeyRangeRef());
+		// DEBUG_KEY_RANGE("makeVersionMutationsDurable", v.version, KeyRangeRef());
 		writeMutations(v.mutations, v.version, "makeVersionDurable");
 		for (const auto& m : v.mutations)
 			bytesLeft -= mvccStorageBytes(m);

flow/network.h

@@ -45,6 +45,7 @@ enum class TaskPriority {
 	WriteSocket = 10000,
 	PollEIO = 9900,
 	DiskIOComplete = 9150,
+	NoDeliverDelay = 9100,
 	LoadBalancedEndpoint = 9000,
 	ReadSocket = 9000,
 	AcceptSocket = 8950,