From ce2d1e1648cbb1eea88990b204be811ae74762ac Mon Sep 17 00:00:00 2001
From: "A.J. Beamon" <aj.beamon@snowflake.com>
Date: Fri, 3 Dec 2021 18:42:55 -0800
Subject: [PATCH] Fix: the shard tracker state could become inaccurate if there
 is an ABA type update to the bandwith state of a shard.

---
 fdbserver/DataDistributionTracker.actor.cpp | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/fdbserver/DataDistributionTracker.actor.cpp b/fdbserver/DataDistributionTracker.actor.cpp
index bafd639ecb..579eb0d6b6 100644
--- a/fdbserver/DataDistributionTracker.actor.cpp
+++ b/fdbserver/DataDistributionTracker.actor.cpp
@@ -434,21 +434,23 @@ ACTOR Future<Void> changeSizes(DataDistributionTracker* self, KeyRange keys, int
 struct HasBeenTrueFor : ReferenceCounted<HasBeenTrueFor> {
 	explicit HasBeenTrueFor(const Optional<ShardMetrics>& value) {
 		if (value.present()) {
-			trigger = delayJittered(std::max(0.0,
-			                                 SERVER_KNOBS->DD_MERGE_COALESCE_DELAY +
-			                                     value.get().lastLowBandwidthStartTime - now()),
-			                        TaskPriority::DataDistributionLow) ||
-			          cleared.getFuture();
+			lowBandwidthStartTime = value.get().lastLowBandwidthStartTime;
+			trigger =
+			    delayJittered(std::max(0.0, SERVER_KNOBS->DD_MERGE_COALESCE_DELAY + lowBandwidthStartTime - now()),
+			                  TaskPriority::DataDistributionLow) ||
+			    cleared.getFuture();
 		}
 	}
 
 	Future<Void> set(double lastLowBandwidthStartTime) {
-		if (!trigger.isValid()) {
+		if (!trigger.isValid() || lowBandwidthStartTime != lastLowBandwidthStartTime) {
 			cleared = Promise<Void>();
 			trigger =
 			    delayJittered(SERVER_KNOBS->DD_MERGE_COALESCE_DELAY + std::max(lastLowBandwidthStartTime - now(), 0.0),
 			                  TaskPriority::DataDistributionLow) ||
 			    cleared.getFuture();
+
+			lowBandwidthStartTime = lastLowBandwidthStartTime;
 		}
 		return trigger;
 	}
@@ -458,12 +460,14 @@ struct HasBeenTrueFor : ReferenceCounted<HasBeenTrueFor> {
 		}
 		trigger = Future<Void>();
 		cleared.send(Void());
+		lowBandwidthStartTime = 0;
 	}
 
 	// True if this->value is true and has been true for this->seconds
 	bool hasBeenTrueForLongEnough() const { return trigger.isValid() && trigger.isReady(); }
 
 private:
+	double lowBandwidthStartTime = 0;
 	Future<Void> trigger;
 	Promise<Void> cleared;
 };