Make ShardArchiveHandler downloads more resilient:

* Make ShardArchiveHandler a singleton.
* Add state database for ShardArchiveHandler.
* Use temporary database for SSLHTTPDownloader downloads.
* Make ShardArchiveHandler a Stoppable class.
* Automatically resume interrupted downloads at server start.

src/ripple/rpc/impl/ShardArchiveHandler.cpp

@@ -22,6 +22,9 @@
 #include <ripple/core/ConfigSections.h>
 #include <ripple/nodestore/DatabaseShard.h>
 #include <ripple/rpc/ShardArchiveHandler.h>
+#include <ripple/basics/BasicConfig.h>
+#include <ripple/rpc/impl/Handler.h>
+#include <ripple/protocol/ErrorCodes.h>
 
 #include <memory>
 
@@ -31,41 +34,192 @@ namespace RPC {
 using namespace boost::filesystem;
 using namespace std::chrono_literals;
 
-ShardArchiveHandler::ShardArchiveHandler(Application& app)
-    : app_(app)
-    , downloadDir_(get(app_.config().section(
-        ConfigSection::shardDatabase()), "path", "") + "/download")
-    , timer_(app_.getIOService())
-    , process_(false)
-    , j_(app.journal("ShardArchiveHandler"))
+std::mutex ShardArchiveHandler::instance_mutex_;
+ShardArchiveHandler::pointer ShardArchiveHandler::instance_ = nullptr;
+
+boost::filesystem::path
+ShardArchiveHandler::getDownloadDirectory(Config const& config)
 {
-    assert(app_.getShardStore());
+    return get(config.section(
+        ConfigSection::shardDatabase()), "download_path",
+            get(config.section(ConfigSection::shardDatabase()),
+                "path", "")) / "download";
 }
 
-ShardArchiveHandler::~ShardArchiveHandler()
+auto
+ShardArchiveHandler::getInstance() -> pointer
 {
-    std::lock_guard lock(m_);
-    timer_.cancel();
-    for (auto const& ar : archives_)
-        app_.getShardStore()->removePreShard(ar.first);
-    archives_.clear();
+    std::lock_guard lock(instance_mutex_);
 
-    // Remove temp root download directory
-    try
-    {
-        remove_all(downloadDir_);
-    }
-    catch (std::exception const& e)
-    {
-        JLOG(j_.error()) <<
-            "exception: " << e.what();
-    }
+    return instance_;
+}
+
+auto
+ShardArchiveHandler::getInstance(Application& app,
+    Stoppable& parent) -> pointer
+{
+    std::lock_guard lock(instance_mutex_);
+    assert(!instance_);
+
+    instance_.reset(new ShardArchiveHandler(app, parent));
+
+    return instance_;
+}
+
+auto
+ShardArchiveHandler::recoverInstance(Application& app, Stoppable& parent) -> pointer
+{
+    std::lock_guard lock(instance_mutex_);
+    assert(!instance_);
+
+    instance_.reset(new ShardArchiveHandler(app, parent, true));
+
+    return instance_;
 }
 
 bool
-ShardArchiveHandler::add(std::uint32_t shardIndex, parsedURL&& url)
+ShardArchiveHandler::hasInstance()
+{
+    std::lock_guard lock(instance_mutex_);
+
+    return instance_.get() != nullptr;
+}
+
+ShardArchiveHandler::ShardArchiveHandler(
+    Application& app,
+    Stoppable& parent,
+    bool recovery)
+    : Stoppable("ShardArchiveHandler", parent)
+    , app_(app)
+    , j_(app.journal("ShardArchiveHandler"))
+    , downloadDir_(getDownloadDirectory(app.config()))
+    , timer_(app_.getIOService())
+    , process_(false)
+{
+    assert(app_.getShardStore());
+
+    if(recovery)
+        downloader_.reset(new DatabaseDownloader (
+            app_.getIOService(), j_, app_.config()));
+}
+
+bool
+ShardArchiveHandler::init()
+{
+    try
+    {
+        create_directories(downloadDir_);
+
+        sqliteDB_ = std::make_unique<DatabaseCon>(
+            downloadDir_ ,
+            stateDBName,
+            DownloaderDBPragma,
+            ShardArchiveHandlerDBInit);
+    }
+    catch(std::exception const& e)
+    {
+        JLOG(j_.error()) << "exception: " << e.what()
+            << " in function: " << __func__;
+
+        return false;
+    }
+
+    return true;
+}
+
+bool
+ShardArchiveHandler::initFromDB()
+{
+    try
+    {
+        using namespace boost::filesystem;
+
+        assert(exists(downloadDir_ / stateDBName) &&
+            is_regular_file(downloadDir_ / stateDBName));
+
+        sqliteDB_ = std::make_unique<DatabaseCon>(
+            downloadDir_,
+            stateDBName,
+            DownloaderDBPragma,
+            ShardArchiveHandlerDBInit);
+
+        auto& session{sqliteDB_->getSession()};
+
+        soci::rowset<soci::row> rs = (session.prepare
+            << "SELECT * FROM State;");
+
+        std::lock_guard<std::mutex> lock(m_);
+
+        for (auto it = rs.begin(); it != rs.end(); ++it)
+        {
+            parsedURL url;
+
+            if (!parseUrl(url, it->get<std::string>(1)))
+            {
+                JLOG(j_.error()) << "Failed to parse url: "
+                    << it->get<std::string>(1);
+
+                continue;
+            }
+
+            add(it->get<int>(0), std::move(url), lock);
+        }
+
+        // Failed to load anything
+        // from the state database.
+        if(archives_.empty())
+        {
+            release();
+            return false;
+        }
+    }
+    catch(std::exception const& e)
+    {
+        JLOG(j_.error()) << "exception: " << e.what()
+            << " in function: " << __func__;
+
+        return false;
+    }
+
+    return true;
+}
+
+void
+ShardArchiveHandler::onStop()
+{
+    std::lock_guard<std::mutex> lock(m_);
+
+    if (downloader_)
+    {
+        downloader_->onStop();
+        downloader_.reset();
+    }
+
+    stopped();
+}
+
+bool
+ShardArchiveHandler::add(std::uint32_t shardIndex,
+    std::pair<parsedURL, std::string>&& url)
+{
+    std::lock_guard<std::mutex> lock(m_);
+
+    if (!add(shardIndex, std::forward<parsedURL>(url.first), lock))
+        return false;
+
+    auto& session{sqliteDB_->getSession()};
+
+    session << "INSERT INTO State VALUES (:index, :url);",
+        soci::use(shardIndex),
+        soci::use(url.second);
+
+    return true;
+}
+
+bool
+ShardArchiveHandler::add(std::uint32_t shardIndex, parsedURL&& url,
+    std::lock_guard<std::mutex> const&)
 {
-    std::lock_guard lock(m_);
     if (process_)
     {
         JLOG(j_.error()) <<
@@ -76,9 +230,12 @@ ShardArchiveHandler::add(std::uint32_t shardIndex, parsedURL&& url)
     auto const it {archives_.find(shardIndex)};
     if (it != archives_.end())
         return url == it->second;
+
     if (!app_.getShardStore()->prepareShard(shardIndex))
         return false;
+
     archives_.emplace(shardIndex, std::move(url));
+
     return true;
 }
 
@@ -107,16 +264,13 @@ ShardArchiveHandler::start()
 
     try
     {
-        // Remove if remnant from a crash
-        remove_all(downloadDir_);
-
         // Create temp root download directory
-        create_directory(downloadDir_);
+        create_directories(downloadDir_);
 
         if (!downloader_)
         {
             // will throw if can't initialize ssl context
-            downloader_ = std::make_shared<SSLHTTPDownloader>(
+            downloader_ = std::make_shared<DatabaseDownloader>(
                 app_.getIOService(), j_, app_.config());
         }
     }
@@ -130,12 +284,19 @@ ShardArchiveHandler::start()
     return next(lock);
 }
 
+void
+ShardArchiveHandler::release()
+{
+    std::lock_guard<std::mutex> lock(m_);
+    doRelease(lock);
+}
+
 bool
 ShardArchiveHandler::next(std::lock_guard<std::mutex>& l)
 {
     if (archives_.empty())
     {
-        process_ = false;
+        doRelease(l);
         return false;
     }
 
@@ -154,20 +315,28 @@ ShardArchiveHandler::next(std::lock_guard<std::mutex>& l)
         return next(l);
     }
 
-    // Download the archive
+    // Download the archive. Process in another thread
+    // to prevent holding up the lock if the downloader
+    // sleeps.
     auto const& url {archives_.begin()->second};
-    if (!downloader_->download(
-        url.domain,
-        std::to_string(url.port.get_value_or(443)),
-        url.path,
-        11,
-        dstDir / "archive.tar.lz4",
-        std::bind(&ShardArchiveHandler::complete,
-            shared_from_this(), std::placeholders::_1)))
-    {
-        remove(l);
-        return next(l);
-    }
+    app_.getJobQueue().addJob(
+        jtCLIENT, "ShardArchiveHandler",
+        [this, ptr = shared_from_this(), url, dstDir](Job&)
+        {
+            if (!downloader_->download(
+                url.domain,
+                std::to_string(url.port.get_value_or(443)),
+                url.path,
+                11,
+                dstDir / "archive.tar.lz4",
+                std::bind(&ShardArchiveHandler::complete,
+                    ptr, std::placeholders::_1)))
+            {
+                std::lock_guard<std::mutex> l(m_);
+                remove(l);
+                next(l);
+            }
+        });
 
     process_ = true;
     return true;
@@ -206,7 +375,7 @@ ShardArchiveHandler::complete(path dstPath)
         jtCLIENT, "ShardArchiveHandler",
         [=, dstPath = std::move(dstPath), ptr = shared_from_this()](Job&)
         {
-            // If validating and not synced then defer and retry
+            // If not synced then defer and retry
             auto const mode {ptr->app_.getOPs().getOperatingMode()};
             if (mode != OperatingMode::FULL)
             {
@@ -282,6 +451,11 @@ ShardArchiveHandler::remove(std::lock_guard<std::mutex>&)
     app_.getShardStore()->removePreShard(shardIndex);
     archives_.erase(shardIndex);
 
+    auto& session{sqliteDB_->getSession()};
+
+    session << "DELETE FROM State WHERE ShardIndex = :index;",
+        soci::use(shardIndex);
+
     auto const dstDir {downloadDir_ / std::to_string(shardIndex)};
     try
     {
@@ -294,5 +468,37 @@ ShardArchiveHandler::remove(std::lock_guard<std::mutex>&)
     }
 }
 
+void
+ShardArchiveHandler::doRelease(std::lock_guard<std::mutex> const&)
+{
+    process_ = false;
+
+    timer_.cancel();
+    for (auto const& ar : archives_)
+        app_.getShardStore()->removePreShard(ar.first);
+    archives_.clear();
+
+    {
+        auto& session{sqliteDB_->getSession()};
+
+        session << "DROP TABLE State;";
+    }
+
+    sqliteDB_.reset();
+
+    // Remove temp root download directory
+    try
+    {
+        remove_all(downloadDir_);
+    }
+    catch (std::exception const& e)
+    {
+        JLOG(j_.error()) << "exception: " << e.what()
+            << " in function: " << __func__;
+    }
+
+    downloader_.reset();
+}
+
 } // RPC
 } // ripple