Merge pull request #2039 from Expensify/dsilva_blockingCommandPortWhe…

…nNotCaughtUp

Blocking command port when node is not caught up

BedrockServer.h

@@ -182,8 +182,8 @@ class BedrockServer : public SQLiteServer {
     void onNodeLogin(SQLitePeer* peer) override;
 
     // You must block and unblock the command port with *identical strings*.
-    void blockCommandPort(const string& reason);
-    void unblockCommandPort(const string& reason);
+    void blockCommandPort(const string& reason) override;
+    void unblockCommandPort(const string& reason) override;
 
     // Legacy version of above.
     void suppressCommandPort(const string& reason, bool suppress, bool manualOverride = false);

sqlitecluster/SQLiteNode.cpp

@@ -1278,7 +1278,23 @@ void SQLiteNode::_onMESSAGE(SQLitePeer* peer, const SData& message) {
             peer->commandAddress = message["commandAddress"];
         }
         peer->setCommit(message.calcU64("CommitCount"), message["Hash"]);
-
+
+        // We check the commit difference with 12,500 commits behind because that 
+        // represents ~30s of commits. If we're behind, let's close the command port
+        // so we can catch up with the cluster before processing new commands.
+        const string blockReason = "COMMITS_LAGGING_BEHIND";
+        const int64_t currentCommitDifference =  peer->commitCount - getCommitCount();
+        if (peer == _leadPeer && currentCommitDifference >= 12'500 && !_blockedCommandPortForBeingBehind) {
+            SINFO("Node is behind by " + SToStr(currentCommitDifference) + " commits, closing command port so it can catch up.");
+            _server.blockCommandPort(blockReason);
+            _blockedCommandPortForBeingBehind = true;
+        } else if (currentCommitDifference < 1'000 && _blockedCommandPortForBeingBehind) {
+            // We'll open the command port again if we're 1k commits behind, which
+            // translates to ~2s of commits.
+            SINFO("Node is caught up enough (behind by " + SToStr(currentCommitDifference) + " commits), re-opening command port.");
+            _server.unblockCommandPort(blockReason);
+            _blockedCommandPortForBeingBehind = false;
+        }
         // Classify and process the message
         if (SIEquals(message.methodLine, "LOGIN")) {
             // LOGIN: This is the first message sent to and received from a new peer. It communicates the current state of

sqlitecluster/SQLiteNode.h

@@ -364,9 +364,12 @@ class SQLiteNode : public STCPManager {
     // Stopwatch to track if we're giving up on the server preventing a standdown.
     SStopwatch _standDownTimeout;
 
-   // Our current State.
+    // Our current State.
     atomic<SQLiteNodeState> _state;
 
+    // Keeps track if we have closed the command port for commits fallen behind.
+    bool _blockedCommandPortForBeingBehind{false};
+
     // This is an integer that increments every time we change states. This is useful for responses to state changes
     // (i.e., approving standup) to verify that the messages we're receiving are relevant to the current state change,
     // and not stale responses to old changes.

sqlitecluster/SQLiteServer.h

@@ -13,4 +13,8 @@ class SQLiteServer : public STCPManager {
 
     // We call this method whenever a node changes state
     virtual void notifyStateChangeToPlugins(SQLite& db, SQLiteNodeState newState) = 0;
+
+    // You must block and unblock the command port with *identical strings*.
+    virtual void blockCommandPort(const string& reason) = 0;
+    virtual void unblockCommandPort(const string& reason) = 0;
 };

test/tests/SQLiteNodeTest.cpp

@@ -26,6 +26,8 @@ class TestServer : public SQLiteServer {
     virtual bool canStandDown() { return true; }
     virtual void onNodeLogin(SQLitePeer* peer) { }
     virtual void notifyStateChangeToPlugins(SQLite& db, SQLiteNodeState newState) {}
+    virtual void blockCommandPort(const string& reason) { };
+    virtual void unblockCommandPort(const string& reason) { };
 };
 
 struct SQLiteNodeTest : tpunit::TestFixture {