Unit tests for database shards

.travis.yml

@@ -408,7 +408,8 @@ cache:
     - $CACHE_DIR
 
 before_install:
-  - if [ "$(uname)" = "Darwin" ] ; then export NUM_PROCESSORS=$(sysctl -n hw.physicalcpu); else export NUM_PROCESSORS=$(nproc); fi
+  # NUM_PROCESSORS was set to 1 due to problems in parallel launch of unit tests on Mac platform
+  - if [ "$(uname)" = "Darwin" ] ; then export NUM_PROCESSORS=1; else export NUM_PROCESSORS=$(nproc); fi
   - echo "NUM PROC is ${NUM_PROCESSORS}"
   - if [ "$(uname)" = "Linux" ] ; then docker pull ${DOCKER_IMAGE}; fi
   - if [ "${MATRIX_EVAL}" != "" ] ; then eval "${MATRIX_EVAL}"; fi

Builds/CMake/RippledCore.cmake

@@ -844,6 +844,7 @@ target_sources (rippled PRIVATE
   #]===============================]
   src/test/nodestore/Backend_test.cpp
   src/test/nodestore/Basics_test.cpp
+  src/test/nodestore/DatabaseShard_test.cpp
   src/test/nodestore/Database_test.cpp
   src/test/nodestore/Timing_test.cpp
   src/test/nodestore/import_test.cpp

Builds/CMake/deps/Findlz4.cmake

@@ -1,6 +1,6 @@
 find_package (PkgConfig)
 if (PKG_CONFIG_FOUND)
-  pkg_search_module (lz4_PC QUIET liblz4>=1.8)
+  pkg_search_module (lz4_PC QUIET liblz4>=1.9)
 endif ()
 
 if(static)

Builds/CMake/deps/Lz4.cmake

@@ -21,7 +21,7 @@ else()
   ExternalProject_Add (lz4
     PREFIX ${nih_cache_path}
     GIT_REPOSITORY https://github.com/lz4/lz4.git
-    GIT_TAG v1.8.2
+    GIT_TAG v1.9.2
     SOURCE_SUBDIR contrib/cmake_unofficial
     CMAKE_ARGS
       -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER}

src/ripple/nodestore/impl/DatabaseShardImp.cpp

@@ -135,13 +135,12 @@ DatabaseShardImp::init()
                     std::make_unique<Shard>(app_, *this, shardIndex, j_)};
                 if (!shard->open(scheduler_, *ctx_))
                 {
-                    if (!shard->isLegacy())
-                        return false;
-
-                    // Remove legacy shard
+                    // Remove corrupted or legacy shard
                     shard->removeOnDestroy();
                     JLOG(j_.warn())
-                        << "shard " << shardIndex << " removed, legacy shard";
+                        << "shard " << shardIndex << " removed, "
+                        << (shard->isLegacy() ? "legacy" : "corrupted")
+                        << " shard";
                     continue;
                 }
 
@@ -276,11 +275,11 @@ DatabaseShardImp::prepareShard(std::uint32_t shardIndex)
     // is greater or equal to the current shard.
     auto seqCheck = [&](std::uint32_t seq) {
         // seq will be greater than zero if valid
-        if (seq > earliestLedgerSeq() && shardIndex >= seqToShardIndex(seq))
+        if (seq >= earliestLedgerSeq() && shardIndex >= seqToShardIndex(seq))
             return fail("has an invalid index");
         return true;
     };
-    if (!seqCheck(app_.getLedgerMaster().getValidLedgerIndex()) ||
+    if (!seqCheck(app_.getLedgerMaster().getValidLedgerIndex() + 1) ||
         !seqCheck(app_.getLedgerMaster().getCurrentLedgerIndex()))
     {
         return false;
@@ -1100,6 +1099,9 @@ DatabaseShardImp::initConfig(std::lock_guard<std::mutex>&)
         ledgersPerShard_ = get<std::uint32_t>(section, "ledgers_per_shard");
         if (ledgersPerShard_ == 0 || ledgersPerShard_ % 256 != 0)
             return fail("'ledgers_per_shard' must be a multiple of 256");
+
+        earliestShardIndex_ = seqToShardIndex(earliestLedgerSeq());
+        avgShardFileSz_ = ledgersPerShard_ * kilobytes(192);
     }
 
     // NuDB is the default and only supported permanent storage backend

src/ripple/nodestore/impl/DatabaseShardImp.h

@@ -234,7 +234,7 @@ class DatabaseShardImp : public DatabaseShard
     std::uint32_t ledgersPerShard_ = ledgersPerShardDefault;
 
     // The earliest shard index
-    std::uint32_t const earliestShardIndex_;
+    std::uint32_t earliestShardIndex_;
 
     // Average storage space required by a shard (in bytes)
     std::uint64_t avgShardFileSz_;

src/ripple/nodestore/impl/codec.h

@@ -51,13 +51,12 @@ lz4_decompress(void const* in, std::size_t in_size, BufferFactory&& bf)
         Throw<std::runtime_error>("lz4 decompress: n == 0");
     void* const out = bf(result.second);
     result.first = out;
-    if (LZ4_decompress_fast(
+    if (LZ4_decompress_safe(
             reinterpret_cast<char const*>(in) + n,
             reinterpret_cast<char*>(out),
-            result.second) +
-            n !=
-        in_size)
-        Throw<std::runtime_error>("lz4 decompress: LZ4_decompress_fast");
+            in_size - n,
+            result.second) != result.second)
+        Throw<std::runtime_error>("lz4 decompress: LZ4_decompress_safe");
     return result;
 }
 

src/test/jtx/Env.h

@@ -325,8 +325,10 @@ class Env
 
             The Application network time is set to
             the close time of the resulting ledger.
+
+        @return true if no error, false if error
     */
-    void
+    bool
     close(
         NetClock::time_point closeTime,
         boost::optional<std::chrono::milliseconds> consensusDelay =
@@ -336,25 +338,29 @@ class Env
 
         The time is calculated as the duration from
         the previous ledger closing time.
+
+        @return true if no error, false if error
     */
     template <class Rep, class Period>
-    void
+    bool
     close(std::chrono::duration<Rep, Period> const& elapsed)
     {
         // VFALCO Is this the correct time?
-        close(now() + elapsed);
+        return close(now() + elapsed);
     }
 
     /** Close and advance the ledger.
 
         The time is calculated as five seconds from
         the previous ledger closing time.
+
+        @return true if no error, false if error
     */
-    void
+    bool
     close()
     {
         // VFALCO Is this the correct time?
-        close(std::chrono::seconds(5));
+        return close(std::chrono::seconds(5));
     }
 
     /** Turn on JSON tracing.

src/test/jtx/impl/Env.cpp

@@ -107,13 +107,14 @@ Env::closed()
     return app().getLedgerMaster().getClosedLedger();
 }
 
-void
+bool
 Env::close(
     NetClock::time_point closeTime,
     boost::optional<std::chrono::milliseconds> consensusDelay)
 {
     // Round up to next distinguishable value
     using namespace std::chrono_literals;
+    bool res = true;
     closeTime += closed()->info().closeTimeResolution - 1s;
     timeKeeper().set(closeTime);
     // Go through the rpc interface unless we need to simulate
@@ -122,10 +123,17 @@ Env::close(
         app().getOPs().acceptLedger(consensusDelay);
     else
     {
-        rpc("ledger_accept");
-        // VFALCO No error check?
+        auto resp = rpc("ledger_accept");
+        if (resp["result"]["status"] != std::string("success"))
+        {
+            JLOG(journal.error())
+                << "Env::close() failed: " << resp["result"]["status"]
+                << std::endl;
+            res = false;
+        }
     }
     timeKeeper().set(closed()->info().closeTime);
+    return res;
 }
 
 void