Drop support for cursors

marta-lokhova · marta-lokhova · commit cafe8792ea4e · 2024-12-17T10:08:17.000-08:00
diff --git a/docs/quick-reference.md b/docs/quick-reference.md
@@ -61,8 +61,7 @@ You have some control over which peers you're connected to:
 
 ### Maintenance
 
-Core keeps old meta around for Horizon and other systems. As cursors get updated, automatic
-maintenance normally deletes more than enough for the node to use a constant amount of disk space.
+Core keeps historical data needed for publish (such as SCP history)
 
 Sometimes you need to clean up more than this (for example, if you have a large maintenance debt).
 In this case running the command `maintenance?count=100000000` (integer is a large number, bigger than your max backlog) will perform the full maintenance.
diff --git a/docs/software/commands.md b/docs/software/commands.md
@@ -244,11 +244,6 @@ Most commands return their results in JSON format.
   `connect?peer=NAME&port=NNN`<br>
   Triggers the instance to connect to peer NAME at port NNN.
 
-* **dropcursor**  
-  `dropcursor?id=ID`<br>
-  Deletes the tracking cursor identified by `id`. See `setcursor` for
-  more information.
-
 * **droppeer**
   `droppeer?node=NODE_ID[&ban=D]`<br>
   Drops peer identified by NODE_ID, when D is 1 the peer is also banned.
@@ -307,23 +302,6 @@ Most commands return their results in JSON format.
   * `delayed`: participating in the latest consensus rounds, but slower than others.
   * `agree`: running just fine.
 
-* **setcursor**
-  `setcursor?id=ID&cursor=N`<br>
-  Sets or creates a cursor identified by `ID` with value `N`. ID is an
-  uppercase AlphaNum, N is an uint32 that represents the last ledger sequence
-  number that the instance ID processed. Cursors are used by dependent services
-  to tell stellar-core which data can be safely deleted by the instance. The
-  data is historical data stored in the SQL tables such as txhistory or
-  ledgerheaders. When all consumers processed the data for ledger sequence N
-  the data can be safely removed by the instance. The actual deletion is
-  performed by invoking the `maintenance` endpoint or on startup. See also
-  `dropcursor`.
-
-* **getcursor**
-  `getcursor?[id=ID]`<br>
-  Gets the cursor identified by `ID`. If ID is not defined then all cursors
-  will be returned.
-
 * **scp**
   `scp?[limit=n][&fullkeys=false]`<br>
   Returns a JSON object with the internal state of the SCP engine for the last
diff --git a/docs/stellar-core_example.cfg b/docs/stellar-core_example.cfg
@@ -304,10 +304,6 @@ KNOWN_PEERS=[
 "core-testnet2.stellar.org",
 "core-testnet3.stellar.org"]
 
-# KNOWN_CURSORS (list of strings) default is empty
-# Set of cursors added at each startup with value '1'.
-KNOWN_CURSORS=["HORIZON"]
-
 #######################
 ##  SCP settings
 
diff --git a/src/bucket/test/BucketManagerTests.cpp b/src/bucket/test/BucketManagerTests.cpp
@@ -22,7 +22,7 @@
 #include "lib/catch.hpp"
 #include "main/Application.h"
 #include "main/Config.h"
-#include "main/ExternalQueue.h"
+#include "main/Maintainer.h"
 #include "test/TestUtils.h"
 #include "test/test.h"
 #include "util/GlobalChecks.h"
@@ -641,8 +641,7 @@ TEST_CASE_VERSIONS(
             clock.crank(false);
 
             // Trim history after publishing whenever possible.
-            ExternalQueue ps(*app);
-            ps.deleteOldEntries(50000);
+            app->getMaintainer().performMaintenance(50000);
         }
     });
 }
diff --git a/src/database/Database.cpp b/src/database/Database.cpp
@@ -24,7 +24,6 @@
 #include "history/HistoryManager.h"
 #include "ledger/LedgerHeaderUtils.h"
 #include "ledger/LedgerTxn.h"
-#include "main/ExternalQueue.h"
 #include "main/PersistentState.h"
 #include "overlay/BanManager.h"
 #include "overlay/OverlayManager.h"
@@ -63,7 +62,7 @@ bool Database::gDriversRegistered = false;
 
 // smallest schema version supported
 static unsigned long const MIN_SCHEMA_VERSION = 21;
-static unsigned long const SCHEMA_VERSION = 23;
+static unsigned long const SCHEMA_VERSION = 24;
 
 // These should always match our compiled version precisely, since we are
 // using a bundled version to get access to carray(). But in case someone
@@ -219,6 +218,9 @@ Database::applySchemaUpgrade(unsigned long vers)
         mApp.getHistoryManager().dropSQLBasedPublish();
         Upgrades::dropSupportUpgradeHistory(*this);
         break;
+    case 24:
+        getSession() << "DROP TABLE IF EXISTS pubsub;";
+        break;
     default:
         throw std::runtime_error("Unknown DB schema version");
     }
@@ -473,7 +475,6 @@ Database::initialize()
     Upgrades::dropAll(*this);
     OverlayManager::dropAll(*this);
     PersistentState::dropAll(*this);
-    ExternalQueue::dropAll(*this);
     LedgerHeaderUtils::dropAll(*this);
     // No need to re-create txhistory, will be dropped during
     // upgradeToCurrentSchema anyway
diff --git a/src/herder/test/HerderTests.cpp b/src/herder/test/HerderTests.cpp
@@ -212,54 +212,6 @@ TEST_CASE_VERSIONS("standalone", "[herder][acceptance]")
                     REQUIRE(c1.loadSequenceNumber() == expectedC1Seq);
                 }
             }
-
-            SECTION("Queue processing test")
-            {
-                app->getCommandHandler().manualCmd("maintenance?queue=true");
-
-                while (app->getLedgerManager().getLastClosedLedgerNum() <
-                       (app->getHistoryManager().getCheckpointFrequency() + 5))
-                {
-                    app->getClock().crank(true);
-                }
-
-                app->getCommandHandler().manualCmd("setcursor?id=A1&cursor=1");
-                app->getCommandHandler().manualCmd("maintenance?queue=true");
-                auto& db = app->getDatabase();
-                auto& sess = db.getSession();
-
-                app->getCommandHandler().manualCmd("setcursor?id=A2&cursor=3");
-                app->getCommandHandler().manualCmd("maintenance?queue=true");
-                auto lh = LedgerHeaderUtils::loadBySequence(db, sess, 2);
-                REQUIRE(!!lh);
-
-                app->getCommandHandler().manualCmd("setcursor?id=A1&cursor=2");
-                // this should delete items older than sequence 2
-                app->getCommandHandler().manualCmd("maintenance?queue=true");
-                lh = LedgerHeaderUtils::loadBySequence(db, sess, 2);
-                REQUIRE(!lh);
-                lh = LedgerHeaderUtils::loadBySequence(db, sess, 3);
-                REQUIRE(!!lh);
-
-                // this should delete items older than sequence 3
-                SECTION("set min to 3 by update")
-                {
-                    app->getCommandHandler().manualCmd(
-                        "setcursor?id=A1&cursor=3");
-                    app->getCommandHandler().manualCmd(
-                        "maintenance?queue=true");
-                    lh = LedgerHeaderUtils::loadBySequence(db, sess, 3);
-                    REQUIRE(!lh);
-                }
-                SECTION("set min to 3 by deletion")
-                {
-                    app->getCommandHandler().manualCmd("dropcursor?id=A1");
-                    app->getCommandHandler().manualCmd(
-                        "maintenance?queue=true");
-                    lh = LedgerHeaderUtils::loadBySequence(db, sess, 3);
-                    REQUIRE(!lh);
-                }
-            }
         }
     });
 }
diff --git a/src/history/test/HistoryTests.cpp b/src/history/test/HistoryTests.cpp
@@ -16,7 +16,7 @@
 #include "historywork/GzipFileWork.h"
 #include "historywork/PutHistoryArchiveStateWork.h"
 #include "ledger/LedgerManager.h"
-#include "main/ExternalQueue.h"
+#include "main/Maintainer.h"
 #include "main/PersistentState.h"
 #include "process/ProcessManager.h"
 #include "test/TestAccount.h"
@@ -1375,8 +1375,7 @@ TEST_CASE("persist publish queue", "[history][publish][acceptance]")
         REQUIRE(hm0.getMinLedgerQueuedToPublish() == 7);
 
         // Trim history after publishing.
-        ExternalQueue ps(*app0);
-        ps.deleteOldEntries(50000);
+        app0->getMaintainer().performMaintenance(50000);
     }
 
     cfg.MAX_CONCURRENT_SUBPROCESSES = 32;
@@ -1395,8 +1394,7 @@ TEST_CASE("persist publish queue", "[history][publish][acceptance]")
             clock.crank(true);
 
             // Trim history after publishing whenever possible.
-            ExternalQueue ps(*app1);
-            ps.deleteOldEntries(50000);
+            app1->getMaintainer().performMaintenance(50000);
         }
         // We should have either an empty publish queue or a
         // ledger sometime after the 5th checkpoint
diff --git a/src/main/ApplicationImpl.cpp b/src/main/ApplicationImpl.cpp
@@ -38,7 +38,6 @@
 #include "main/AppConnector.h"
 #include "main/ApplicationUtils.h"
 #include "main/CommandHandler.h"
-#include "main/ExternalQueue.h"
 #include "main/Maintainer.h"
 #include "main/StellarCoreVersion.h"
 #include "medida/counter.h"
@@ -872,9 +871,6 @@ ApplicationImpl::startServices()
 {
     // restores Herder's state before starting overlay
     mHerder->start();
-    // set known cursors before starting maintenance job
-    ExternalQueue ps(*this);
-    ps.setInitialCursors(mConfig.KNOWN_CURSORS);
     mMaintainer->start();
     if (mConfig.MODE_AUTO_STARTS_OVERLAY)
     {
diff --git a/src/main/CommandHandler.cpp b/src/main/CommandHandler.cpp
@@ -36,8 +36,6 @@
 #include "xdr/Stellar-transaction.h"
 #include "xdrpp/marshal.h"
 
-#include "ExternalQueue.h"
-
 #ifdef BUILD_TESTS
 #include "simulation/LoadGenerator.h"
 #include "test/TestAccount.h"
@@ -90,9 +88,6 @@ CommandHandler::CommandHandler(Application& app) : mApp(app)
     mServer->add404(std::bind(&CommandHandler::fileNotFound, this, _1, _2));
     if (mApp.getConfig().modeStoresAnyHistory())
     {
-        addRoute("dropcursor", &CommandHandler::dropcursor);
-        addRoute("getcursor", &CommandHandler::getcursor);
-        addRoute("setcursor", &CommandHandler::setcursor);
         addRoute("maintenance", &CommandHandler::maintenance);
     }
 
@@ -1011,77 +1006,6 @@ CommandHandler::tx(std::string const& params, std::string& retStr)
     retStr = Json::FastWriter().write(root);
 }
 
-void
-CommandHandler::dropcursor(std::string const& params, std::string& retStr)
-{
-    ZoneScoped;
-    std::map<std::string, std::string> map;
-    http::server::server::parseParams(params, map);
-    std::string const& id = map["id"];
-
-    if (!ExternalQueue::validateResourceID(id))
-    {
-        retStr = "Invalid resource id";
-    }
-    else
-    {
-        ExternalQueue ps(mApp);
-        ps.deleteCursor(id);
-        retStr = "Done";
-    }
-}
-
-void
-CommandHandler::setcursor(std::string const& params, std::string& retStr)
-{
-    ZoneScoped;
-    std::map<std::string, std::string> map;
-    http::server::server::parseParams(params, map);
-    std::string const& id = map["id"];
-
-    uint32 cursor = parseRequiredParam<uint32>(map, "cursor");
-
-    if (!ExternalQueue::validateResourceID(id))
-    {
-        retStr = "Invalid resource id";
-    }
-    else
-    {
-        ExternalQueue ps(mApp);
-        ps.setCursorForResource(id, cursor);
-        retStr = "Done";
-    }
-}
-
-void
-CommandHandler::getcursor(std::string const& params, std::string& retStr)
-{
-    ZoneScoped;
-    Json::Value root;
-    std::map<std::string, std::string> map;
-    http::server::server::parseParams(params, map);
-    std::string const& id = map["id"];
-
-    // the decision was made not to check validity here
-    // because there are subsequent checks for that in
-    // ExternalQueue and if an exception is thrown for
-    // validity there, the ret format is technically more
-    // correct for the mime type
-    ExternalQueue ps(mApp);
-    std::map<std::string, uint32> curMap;
-    int counter = 0;
-    ps.getCursorForResource(id, curMap);
-    root["cursors"][0];
-    for (auto cursor : curMap)
-    {
-        root["cursors"][counter]["id"] = cursor.first;
-        root["cursors"][counter]["cursor"] = cursor.second;
-        counter++;
-    }
-
-    retStr = root.toStyledString();
-}
-
 void
 CommandHandler::maintenance(std::string const& params, std::string& retStr)
 {
diff --git a/src/main/CommandHandler.h b/src/main/CommandHandler.h
@@ -49,7 +49,6 @@ class CommandHandler
 
     void bans(std::string const& params, std::string& retStr);
     void connect(std::string const& params, std::string& retStr);
-    void dropcursor(std::string const& params, std::string& retStr);
     void dropPeer(std::string const& params, std::string& retStr);
     void info(std::string const& params, std::string& retStr);
     void ll(std::string const& params, std::string& retStr);
@@ -61,8 +60,6 @@ class CommandHandler
     void peers(std::string const& params, std::string& retStr);
     void selfCheck(std::string const&, std::string& retStr);
     void quorum(std::string const& params, std::string& retStr);
-    void setcursor(std::string const& params, std::string& retStr);
-    void getcursor(std::string const& params, std::string& retStr);
     void scpInfo(std::string const& params, std::string& retStr);
     void tx(std::string const& params, std::string& retStr);
     void unban(std::string const& params, std::string& retStr);
diff --git a/src/main/Config.cpp b/src/main/Config.cpp
@@ -8,7 +8,6 @@
 #include "herder/Herder.h"
 #include "history/HistoryArchive.h"
 #include "ledger/LedgerManager.h"
-#include "main/ExternalQueue.h"
 #include "main/StellarCoreVersion.h"
 #include "scp/LocalNode.h"
 #include "scp/QuorumSetUtils.h"
@@ -1127,18 +1126,6 @@ Config::processConfig(std::shared_ptr<cpptoml::table> t)
                  [&]() { BUCKETLIST_DB_PERSIST_INDEX = readBool(item); }},
                 {"METADATA_DEBUG_LEDGERS",
                  [&]() { METADATA_DEBUG_LEDGERS = readInt<uint32_t>(item); }},
-                {"KNOWN_CURSORS",
-                 [&]() {
-                     KNOWN_CURSORS = readArray<std::string>(item);
-                     for (auto const& c : KNOWN_CURSORS)
-                     {
-                         if (!ExternalQueue::validateResourceID(c))
-                         {
-                             throw std::invalid_argument(fmt::format(
-                                 FMT_STRING("invalid cursor: \"{}\""), c));
-                         }
-                     }
-                 }},
                 {"RUN_STANDALONE", [&]() { RUN_STANDALONE = readBool(item); }},
                 {"CATCHUP_COMPLETE",
                  [&]() { CATCHUP_COMPLETE = readBool(item); }},
diff --git a/src/main/Config.h b/src/main/Config.h
@@ -554,9 +554,6 @@ class Config : public std::enable_shared_from_this<Config>
     // at a premium.
     uint32_t METADATA_DEBUG_LEDGERS;
 
-    // Set of cursors added at each startup with value '1'.
-    std::vector<std::string> KNOWN_CURSORS;
-
     // maximum protocol version supported by the application, can be overridden
     // in tests
     uint32_t LEDGER_PROTOCOL_VERSION;
diff --git a/src/main/ExternalQueue.cpp b/src/main/ExternalQueue.cpp
diff --git a/src/main/ExternalQueue.h b/src/main/ExternalQueue.h
diff --git a/src/main/Maintainer.cpp b/src/main/Maintainer.cpp
diff --git a/src/main/test/ExternalQueueTests.cpp b/src/main/test/ExternalQueueTests.cpp
