Implement pause/resume interface for shard downloader.

Add unit tests.

Builds/CMake/RippledCore.cmake

@@ -555,6 +555,8 @@ else ()
        nounity, main sources:
          subdir: net
     #]===============================]
+    src/ripple/net/impl/DatabaseDownloader.cpp
+    src/ripple/net/impl/FileDownloader.cpp
     src/ripple/net/impl/HTTPClient.cpp
     src/ripple/net/impl/InfoSub.cpp
     src/ripple/net/impl/RPCCall.cpp
@@ -978,6 +980,7 @@ else ()
     src/test/rpc/RPCOverload_test.cpp
     src/test/rpc/RobustTransaction_test.cpp
     src/test/rpc/ServerInfo_test.cpp
+    src/test/rpc/ShardArchiveHandler_test.cpp
     src/test/rpc/Status_test.cpp
     src/test/rpc/Submit_test.cpp
     src/test/rpc/Subscribe_test.cpp

src/ripple/app/main/Application.cpp

@@ -64,6 +64,7 @@
 #include <ripple/rpc/impl/RPCHelpers.h>
 #include <ripple/beast/asio/io_latency_probe.h>
 #include <ripple/beast/core/LexicalCast.h>
+#include <ripple/rpc/ShardArchiveHandler.h>
 
 #include <boost/algorithm/string/predicate.hpp>
 #include <ripple/app/main/GRPCServer.h>
@@ -1607,6 +1608,19 @@ bool ApplicationImp::setup()
         }
     }
 
+    if (shardStore_)
+    {
+        using namespace boost::filesystem;
+
+        auto stateDb(
+            RPC::ShardArchiveHandler::getStateDirectory(*config_)
+            / "stateDatabase");
+
+        if(exists(stateDb) && is_regular_file(stateDb) &&
+           !RPC::ShardArchiveHandler::hasInstance())
+            RPC::ShardArchiveHandler::getInstance(*this);
+    }
+
     return true;
 }
 

src/ripple/core/DatabaseCon.h

@@ -102,18 +102,17 @@ class DatabaseCon
         boost::filesystem::path pPath =
             useTempFiles ? "" : (setup.dataDir / DBName);
 
-        open(session_, "sqlite", pPath.string());
+        init(pPath, pragma, initSQL);
+    }
 
-        for (auto const& p : pragma)
-        {
-            soci::statement st = session_.prepare << p;
-            st.execute(true);
-        }
-        for (auto const& sql : initSQL)
-        {
-            soci::statement st = session_.prepare << sql;
-            st.execute(true);
-        }
+    template<std::size_t N, std::size_t M>
+    DatabaseCon(
+        boost::filesystem::path const& dataDir,
+        std::string const& DBName,
+        std::array<char const*, N> const& pragma,
+        std::array<char const*, M> const& initSQL)
+    {
+        init((dataDir / DBName), pragma, initSQL);
     }
 
     soci::session& getSession()
@@ -129,6 +128,27 @@ class DatabaseCon
     void setupCheckpointing (JobQueue*, Logs&);
 
 private:
+
+    template<std::size_t N, std::size_t M>
+    void
+    init(boost::filesystem::path const& pPath,
+        std::array<char const*, N> const& pragma,
+        std::array<char const*, M> const& initSQL)
+    {
+        open(session_, "sqlite", pPath.string());
+
+        for (auto const& p : pragma)
+        {
+            soci::statement st = session_.prepare << p;
+            st.execute(true);
+        }
+        for (auto const& sql : initSQL)
+        {
+            soci::statement st = session_.prepare << sql;
+            st.execute(true);
+        }
+    }
+
     LockedSociSession::mutex lock_;
 
     soci::session session_;

src/ripple/net/DatabaseBody.h

@@ -0,0 +1,191 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2012, 2018 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#ifndef RIPPLE_NET_DATABASEBODY_H
+#define RIPPLE_NET_DATABASEBODY_H
+
+#include <ripple/core/DatabaseCon.h>
+#include <soci/sqlite3/soci-sqlite3.h>
+#include <boost/beast/http/message.hpp>
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/spawn.hpp>
+
+namespace ripple
+{
+
+struct DatabaseBody
+{
+    // Algorithm for storing buffers when parsing.
+    class reader;
+
+    // The type of the @ref message::body member.
+    class value_type;
+
+    /** Returns the size of the body
+
+        @param body The database body to use
+    */
+    static
+    std::uint64_t
+    size(value_type const& body);
+};
+
+class DatabaseBody::value_type
+{
+    // This body container holds a connection to the
+    // database, and also caches the size when set.
+
+    friend class reader;
+    friend struct DatabaseBody;
+
+    static constexpr
+    std::array<char const*, 2> DatabasePragma
+    {{
+         "PRAGMA synchronous=FULL;",
+         "PRAGMA journal_mode=DELETE;"
+     }};
+
+    static constexpr
+    std::array<char const*, 3> DatabaseInit
+    {{
+        "BEGIN TRANSACTION;",
+
+        "CREATE TABLE IF NOT EXISTS download (       \
+            Path        TEXT,                        \
+            Data        BLOB,                        \
+            Size        BIGINT UNSIGNED,             \
+            Part        BIGINT UNSIGNED PRIMARY KEY  \
+        );",
+
+        "END TRANSACTION;"
+     }};
+
+    // The cached file size
+    std::uint64_t file_size_ = 0;
+    boost::filesystem::path path_;
+    std::unique_ptr<DatabaseCon> conn_;
+    std::string batch_;
+    std::shared_ptr<boost::asio::io_service::strand> strand_;
+    std::mutex m_;
+    std::condition_variable c_;
+    uint64_t handler_count_ = 0;
+    uint64_t part_ = 0;
+
+public:
+
+    /// Destructor
+    ~value_type() = default;
+
+    /// Constructor
+    value_type() = default;
+
+    /// Returns `true` if the file is open
+    bool
+    is_open() const
+    {
+        return bool {conn_ };
+    }
+
+    /// Returns the size of the file if open
+    std::uint64_t
+    size() const
+    {
+        return file_size_;
+    }
+
+    /// Close the file if open
+    void
+    close();
+
+    /** Open a file at the given path with the specified mode
+
+        @param path The utf-8 encoded path to the file
+
+        @param mode The file mode to use
+
+        @param ec Set to the error, if any occurred
+    */
+    void
+    open(boost::filesystem::path path, Config const& config,
+        boost::asio::io_service& io_service, boost::system::error_code & ec);
+};
+
+/** Algorithm for storing buffers when parsing.
+
+    Objects of this type are created during parsing
+    to store incoming buffers representing the body.
+*/
+class DatabaseBody::reader
+{
+    value_type& body_;  // The body we are writing to
+
+    static const uint32_t FLUSH_SIZE   = 50000000;
+    static const uint8_t  MAX_HANDLERS = 3;
+    static const uint16_t MAX_ROW_SIZE_PAD = 500;
+
+public:
+    // Constructor.
+    //
+    // This is called after the header is parsed and
+    // indicates that a non-zero sized body may be present.
+    // `h` holds the received message headers.
+    // `b` is an instance of `DatabaseBody`.
+    //
+    template<bool isRequest, class Fields>
+    explicit
+    reader(boost::beast::http::header<isRequest, Fields>&h, value_type& b);
+
+    // Initializer
+    //
+    // This is called before the body is parsed and
+    // gives the reader a chance to do something that might
+    // need to return an error code. It informs us of
+    // the payload size (`content_length`) which we can
+    // optionally use for optimization.
+    //
+    void
+    init(boost::optional<std::uint64_t> const&, boost::system::error_code& ec);
+
+    // This function is called one or more times to store
+    // buffer sequences corresponding to the incoming body.
+    //
+    template<class ConstBufferSequence>
+    std::size_t
+    put(ConstBufferSequence const& buffers,
+        boost::system::error_code& ec);
+
+    void
+    do_put(std::string data);
+
+    // This function is called when writing is complete.
+    // It is an opportunity to perform any final actions
+    // which might fail, in order to return an error code.
+    // Operations that might fail should not be attempted in
+    // destructors, since an exception thrown from there
+    // would terminate the program.
+    //
+    void
+    finish(boost::system::error_code& ec);
+};
+
+} // ripple
+
+#include <ripple/net/impl/DatabaseBody.ipp>
+
+#endif //RIPPLE_NET_DATABASEBODY_H

src/ripple/net/DatabaseDownloader.h

@@ -0,0 +1,62 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2012, 2018 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#ifndef RIPPLE_NET_DATABASEDOWNLOADER_H
+#define RIPPLE_NET_DATABASEDOWNLOADER_H
+
+#include <ripple/net/SSLHTTPDownloader.h>
+#include <ripple/net/DatabaseBody.h>
+
+namespace ripple
+{
+
+class DatabaseDownloader : public SSLHTTPDownloader
+{
+public:
+
+    DatabaseDownloader(
+        boost::asio::io_service& io_service,
+        beast::Journal j,
+        Config const& config);
+
+private:
+
+    static const uint8_t MAX_PATH_LEN = std::numeric_limits<uint8_t>::max();
+
+    std::shared_ptr<parser>
+    getParser(boost::filesystem::path dstPath,
+        std::function<void(boost::filesystem::path)> complete,
+        boost::system::error_code & ec) override;
+
+    bool
+    checkPath(boost::filesystem::path const& dstPath) override;
+
+    void
+    closeBody(std::shared_ptr<parser> p) override;
+
+    uint64_t
+    size(std::shared_ptr<parser> p) override;
+
+    Config const& config_;
+    boost::asio::io_service & io_service_;
+};
+
+} // ripple
+
+#endif //RIPPLE_NET_DATABASEDOWNLOADER_H