Log duplicate escalated write requests

BedrockServer.cpp

@@ -1839,6 +1839,7 @@ void BedrockServer::_status(unique_ptr<BedrockCommand>& command)
         content["host"] = args["-nodeHost"];
         content["commandCount"] = BedrockCommand::getCommandCount();
         content["isDetached"] = isDetached() ? "true" : "false";
+        content["duplicateEscalatedRequestCount"] = _duplicateEscalatedRequestCount.load();
 
         {
             // Make it known if anything is known to cause crashes.
@@ -2458,6 +2459,50 @@ string BedrockServer::generateCommandID()
     return args["-nodeName"] + "#" + to_string(_requestCount++);
 }
 
+size_t BedrockServer::escalatedRequestFingerprint(const SData& request)
+{
+    // Headers that can be different between duplicate requests from a client
+    static const set<string, STableComp> excludedHeaders = {
+        "ID", "timeout", "serializedData", "httpsRequests", "Content-Length",
+    };
+
+    string canonical = request.methodLine;
+    canonical += '\n';
+    for (const auto& [name, value] : request.nameValueMap) {
+        if (excludedHeaders.count(name)) {
+            continue;
+        }
+        canonical += name;
+        canonical += ':';
+        canonical += value;
+        canonical += '\n';
+    }
+    canonical += request.content;
+    return hash<string>{}(canonical);
+}
+
+size_t BedrockServer::trackInFlightWrite(const SData& request)
+{
+    const size_t fingerprint = escalatedRequestFingerprint(request);
+    bool isDuplicate = false;
+    {
+        lock_guard<mutex> lock(_inFlightEscalatedRequestsMutex);
+        isDuplicate = !_inFlightEscalatedRequests.insert(fingerprint).second;
+    }
+    if (isDuplicate) {
+        _duplicateEscalatedRequestCount++;
+        SWARN("Duplicate write in flight on leader. command:" << request.methodLine
+              << ", requestID:" << request["requestID"] << ", fingerprint:" << fingerprint);
+    }
+    return fingerprint;
+}
+
+void BedrockServer::untrackInFlightWrite(size_t fingerprint)
+{
+    lock_guard<mutex> lock(_inFlightEscalatedRequestsMutex);
+    _inFlightEscalatedRequests.erase(fingerprint);
+}
+
 unique_ptr<BedrockCommand> BedrockServer::buildCommandFromRequest(SData&& request, Socket& socket, bool shouldTreatAsLocalhost)
 {
     SAUTOPREFIX(request);
@@ -2635,6 +2680,16 @@ void BedrockServer::handleSocket(Socket&& socket, bool fromControlPort, bool fro
                             command->response.methodLine = "500 Server Shutting Down";
                             _reply(command);
                         } else {
+                            // Detect duplicate writes reaching this leader: the same write arriving again while a
+                            // previous copy is still being processed, causing write amplification. A duplicate can
+                            // arrive escalated from a follower (private command port) or sent directly by a client that
+                            // connected to the leader (public command port).
+                            size_t inFlightFingerprint = 0;
+                            const bool trackInFlight = fromPrivateCommandPort || (fromPublicCommandPort && getState() == SQLiteNodeState::LEADING);
+                            if (trackInFlight) {
+                                inFlightFingerprint = trackInFlightWrite(command->request);
+                            }
+
                             // If it's not handled by `_handleIfStatusOrControlCommand` we fall into the queuing logic.
                             // If the command has a socket (it's this socket) then we need to wait for it to finish before
                             // we can dequeue the next command, so that the responses all end up delivered in order.
@@ -2692,6 +2747,11 @@ void BedrockServer::handleSocket(Socket&& socket, bool fromControlPort, bool fro
                             }
 
                             commandThread.join();
+
+                            // The write is done, stop tracking it.
+                            if (trackInFlight) {
+                                untrackInFlightWrite(inFlightFingerprint);
+                            }
                         }
                     }
                 }

BedrockServer.h

@@ -373,6 +373,15 @@ class BedrockServer : public SQLiteServer {
     multimap<uint64_t, uint64_t> _futureCommitCommandTimeouts;
     recursive_mutex _futureCommitCommandMutex;
 
+    // Tracks a content fingerprint for each escalated request currently in flight on the leader. It's used to detect a
+    // duplicate escalated request that arrives while the same request is still being processed.
+    set<size_t> _inFlightEscalatedRequests;
+    mutex _inFlightEscalatedRequestsMutex;
+
+    // A cumulative count of the duplicate escalated requests detected since startup, reported in the `Status` command.
+    // This is mostly useful to enable automated tests on this approach.
+    atomic<uint64_t> _duplicateEscalatedRequestCount{0};
+
     // A set of command names that will always be run with QUORUM consistency level.
     // Specified by the `-synchronousCommands` command-line switch.
     set<string> _syncCommands;
@@ -457,6 +466,16 @@ class BedrockServer : public SQLiteServer {
     // Setup a new command from a bare request.
     unique_ptr<BedrockCommand> buildCommandFromRequest(SData&& request, Socket& s, bool shouldTreatAsLocalhost);
 
+    // Computes a content fingerprint for a request, used to detect duplicate writes on the leader. It hashes the
+    // method line, the request headers, and the body, but skips the transport headers a follower stamps on while
+    // escalating, which differ between two copies of the same write.
+    static size_t escalatedRequestFingerprint(const SData& request);
+
+    // Tracks/untracks an in-flight write on the leader by content fingerprint, so we can detect a duplicate copy
+    // arriving while the first is still being processed.
+    size_t trackInFlightWrite(const SData& request);
+    void untrackInFlightWrite(size_t fingerprint);
+
     // This is a monotonically incrementing integer just used to uniquely identify socket threads.
     atomic<uint64_t> _socketThreadNumber;
 

test/clustertest/tests/EscalateTest.cpp

@@ -9,7 +9,8 @@ struct EscalateTest : tpunit::TestFixture
                                          TEST(EscalateTest::test),
                                          TEST(EscalateTest::testSerializedData),
                                          TEST(EscalateTest::testSerializedDataException),
-                                         TEST(EscalateTest::socketReuse))
+                                         TEST(EscalateTest::socketReuse),
+                                         TEST(EscalateTest::duplicateEscalatedRequestDetected))
     {
     }
 
@@ -106,4 +107,34 @@ struct EscalateTest : tpunit::TestFixture
         results = brtester.executeWaitMultipleData({cmd});
         ASSERT_EQUAL(results[0].methodLine, "200 OK");
     }
+
+    void duplicateEscalatedRequestDetected()
+    {
+        // Node 0 is the leader (which tracks duplicates), node 1 a follower (which we escalate from).
+        BedrockTester& leader = tester->getTester(0);
+        BedrockTester& follower = tester->getTester(1);
+        ASSERT_TRUE(leader.waitForState("LEADING"));
+        ASSERT_TRUE(follower.waitForState("FOLLOWING"));
+
+        // Read the leader's current duplicate count so the assertion is independent of any earlier tests.
+        uint64_t countBefore = SToUInt64(SParseJSONObject(leader.executeWaitVerifyContent(SData("Status"), "200", true))["duplicateEscalatedRequestCount"]);
+
+        // Two identical writes sharing one `requestID`. Their command ids differ, but the content fingerprint matches.
+        // The slow process step keeps the first copy in flight long enough for the second to arrive while the first is still being processed.
+        SData cmd("idcollision");
+        cmd["writeConsistency"] = "ASYNC";
+        cmd["requestID"] = "duplicate_escalation_test";
+        cmd["ProcessSleep"] = "1000";
+        cmd["value"] = "duplicate";
+
+        // Send both at once (one per connection) so they escalate to the leader concurrently.
+        auto results = follower.executeWaitMultipleData({cmd, cmd}, 2);
+        ASSERT_EQUAL(results.size(), 2);
+        ASSERT_EQUAL(results[0].methodLine, "200 OK");
+        ASSERT_EQUAL(results[1].methodLine, "200 OK");
+
+        // The leader should have detected at least one duplicate.
+        uint64_t countAfter = SToUInt64(SParseJSONObject(leader.executeWaitVerifyContent(SData("Status"), "200", true))["duplicateEscalatedRequestCount"]);
+        ASSERT_GREATER_THAN(countAfter, countBefore);
+    }
 } __EscalateTest;