Merge branch 'dev' into catalogue_fixes

Builds/levelization/results/ordering.txt

@@ -127,6 +127,7 @@ test.rpc > xrpl.basics
 test.rpc > xrpld.app
 test.rpc > xrpld.core
 test.rpc > xrpld.net
+test.rpc > xrpld.nodestore
 test.rpc > xrpld.overlay
 test.rpc > xrpld.rpc
 test.rpc > xrpld.shamap

include/xrpl/protocol/jss.h

@@ -171,6 +171,7 @@ JSS(amendment_blocked);           // out: NetworkOPs
 JSS(amm_account);                 // in: amm_info
 JSS(amount);                      // out: AccountChannels, amm_info
 JSS(amount2);                     // out: amm_info
+JSS(anchored_through);            // out: CatalogueLoad
 JSS(api_version);                 // in: many, out: Version
 JSS(api_version_low);             // out: Version
 JSS(applied);                     // out: SubmitTransaction
@@ -744,6 +745,7 @@ JSS(TRANSACTION_FLAGS_INDICES);  // out: RPC server_definitions
 JSS(type_hex);                   // out: STPathSet
 JSS(unl);                        // out: UnlList
 JSS(unlimited);                  // out: Connection.h
+JSS(unpinned);                   // out: CatalogueUnpin
 JSS(uptime);                     // out: GetCounts
 JSS(uri);                        // out: ValidatorSites
 JSS(url);                        // in/out: Subscribe, Unsubscribe

src/test/app/SHAMapStore_test.cpp

@@ -33,9 +33,11 @@
 #include <xrpl/protocol/jss.h>
 #include <boost/filesystem.hpp>
 
+#include <chrono>
 #include <exception>
 #include <limits>
 #include <string>
+#include <thread>
 
 namespace ripple {
 namespace test {
@@ -93,6 +95,27 @@ class SHAMapStore_test : public beast::unit_test::suite
         return cfg;
     }
 
+    // Online delete with DatabasePinned over in-memory backends. Unlike
+    // pinnedPersistent, the hot store keeps nodes (type=rwdb is the null
+    // node store), so ledgers closed in the test can be fetched. There is
+    // no database_path, so every SHAMapStoreImp opens its own private
+    // state.db: a second instance built by a test does not see (or trip
+    // over) the rotation state of the Env's own.
+    static auto
+    pinnedInMemory(
+        std::unique_ptr<Config> cfg,
+        std::string const& nodePath,
+        std::string const& pinnedPath)
+    {
+        cfg = onlineDelete(std::move(cfg));
+        auto& section = cfg->section(ConfigSection::nodeDatabase());
+        section.set("type", "memory");
+        section.set("path", nodePath);
+        section.set("pinned_type", "memory");
+        section.set("pinned_path", pinnedPath);
+        return cfg;
+    }
+
     bool
     goodLedger(
         jtx::Env& env,
@@ -1153,6 +1176,127 @@ public:
         }
     }
 
+    // catalogue_unpin of everything (or a rolled-back first load) leaves the
+    // pinned set empty. That must persist as "no pins", not as a string that
+    // fails to parse on the next start.
+    void
+    testPinnedRangeRestoreEmpty()
+    {
+        testcase("pinned range restore of an empty set");
+
+        using namespace jtx;
+
+        beast::temp_dir tempDir;
+        auto const tempPath = boost::filesystem::path(tempDir.path());
+        auto const nodePath = (tempPath / "node").string();
+        auto const pinnedPath = (tempPath / "pinned").string();
+
+        Env env(*this, envconfig(pinnedInMemory, nodePath, pinnedPath));
+
+        NodeStoreScheduler scheduler(env.app().getJobQueue());
+        auto const journal = env.app().journal("SHAMapStoreTest");
+        try
+        {
+            SHAMapStoreImp restoreCheck(env.app(), scheduler, journal);
+
+            RangeSet<std::uint32_t> ranges;
+            ranges.insert(range(100u, 200u));
+            restoreCheck.setPinnedRanges(ranges);
+            restoreCheck.setPinnedRanges(RangeSet<std::uint32_t>{});
+
+            restoreCheck.start();
+            restoreCheck.stop();
+            pass();
+        }
+        catch (std::exception const& e)
+        {
+            fail(
+                std::string("restart with no pinned ranges threw: ") +
+                e.what());
+        }
+        BEAST_EXPECT(
+            env.app().getLedgerMaster().getPinnedLedgersRangeSet().empty());
+    }
+
+    // A crash between a pinned-range checkpoint and the backend committing
+    // the last ledgers leaves the recorded interval's tail missing. Startup
+    // trims it back to the last ledger present instead of refusing to start.
+    void
+    testPinnedRangeRestoreTrimsMissingTail()
+    {
+        testcase("pinned range restore trims a missing tail");
+
+        using namespace jtx;
+
+        beast::temp_dir tempDir;
+        auto const tempPath = boost::filesystem::path(tempDir.path());
+        auto const nodePath = (tempPath / "node").string();
+        auto const pinnedPath = (tempPath / "pinned").string();
+
+        Env env(*this, envconfig(pinnedInMemory, nodePath, pinnedPath));
+        for (int i = 0; i < 4; ++i)
+            env.close();
+
+        auto& rdb = env.app().getRelationalDatabase();
+        auto const last = env.app().getLedgerMaster().getValidLedgerIndex();
+        auto const first = last - 2;
+
+        // Ledger saves may complete asynchronously; wait for the SQL rows.
+        for (int i = 0; i < 100 && rdb.getHashByIndex(last).isZero(); ++i)
+            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+        if (!BEAST_EXPECT(
+                rdb.getHashByIndex(first).isNonZero() &&
+                rdb.getHashByIndex(last).isNonZero()))
+            return;
+
+        NodeStoreScheduler scheduler(env.app().getJobQueue());
+        auto const journal = env.app().journal("SHAMapStoreTest");
+        try
+        {
+            SHAMapStoreImp restoreCheck(env.app(), scheduler, journal);
+
+            // Recorded through last + 10, but only [first, last] exist.
+            RangeSet<std::uint32_t> recorded;
+            recorded.insert(range(first, last + 10));
+            restoreCheck.setPinnedRanges(recorded);
+
+            restoreCheck.start();
+            restoreCheck.stop();
+        }
+        catch (std::exception const& e)
+        {
+            fail(std::string("restart with a missing tail threw: ") + e.what());
+            return;
+        }
+
+        RangeSet<std::uint32_t> expected;
+        expected.insert(range(first, last));
+        BEAST_EXPECT(
+            env.app().getLedgerMaster().getPinnedLedgersRangeSet() == expected);
+
+        // A missing tail longer than the trim limit (65536 ledgers) is not
+        // an interrupted load: it means the wrong pinned store, so startup
+        // still refuses.
+        try
+        {
+            SHAMapStoreImp restoreCheck(env.app(), scheduler, journal);
+
+            RangeSet<std::uint32_t> recorded;
+            recorded.insert(range(first, last + 70000));
+            restoreCheck.setPinnedRanges(recorded);
+
+            restoreCheck.start();
+            restoreCheck.stop();
+            fail("Expected startup failure for an over-long missing tail");
+        }
+        catch (std::runtime_error const& e)
+        {
+            BEAST_EXPECT(
+                std::string(e.what()).find("Persisted pinned interval") !=
+                std::string::npos);
+        }
+    }
+
     // Helper: compare two RangeSet<uint32_t> for set-equality. The
     // test cares about which sequences are in the result, not the
     // particular interval representation.
@@ -1302,6 +1446,8 @@ public:
         testNullFactoryDropsWrites();
         testRotate();
         testPinnedRangeRestoreRequiresPinnedData();
+        testPinnedRangeRestoreEmpty();
+        testPinnedRangeRestoreTrimsMissingTail();
     }
 };
 

src/test/nodestore/Backend_test.cpp

@@ -66,6 +66,14 @@ public:
             // Write the batch
             storeBatch(*backend, batch);
 
+            // Durability on demand: RocksDB drains its write queue and
+            // syncs its WAL; in-memory backends have nothing to wait for;
+            // NuDB commits on its own thread and cannot promise it.
+            backend->store(batch.front());
+            bool const durable = backend->syncDurable();
+            BEAST_EXPECT(durable == (type != "nudb"));
+            backend->sync();
+
             {
                 // Read it back in
                 Batch copy;

src/test/rpc/Catalogue_test.cpp

@@ -18,16 +18,22 @@
 //==============================================================================
 
 #include <test/jtx.h>
+#include <xrpld/app/ledger/Ledger.h>
 #include <xrpld/app/ledger/LedgerMaster.h>
 #include <xrpld/core/ConfigSections.h>
+#include <xrpld/nodestore/detail/DatabasePinnedImp.h>
 #include <xrpl/beast/unit_test.h>
 #include <xrpl/beast/utility/temp_dir.h>
 #include <xrpl/protocol/jss.h>
 #include <boost/filesystem.hpp>
 #include <chrono>
+#include <cstddef>
 #include <cstdlib>
+#include <cstring>
 #include <fstream>
 #include <grpc/impl/codegen/compression_types.h>
+#include <iterator>
+#include <optional>
 #include <thread>
 
 namespace ripple {
@@ -102,14 +108,28 @@ struct CatalogueEnvOpts
 };
 
 // envconfig with pinned_type set so catalogue_load passes the config guard.
-// Uses rwdb for in-process speed (allowed in standalone).
+//
+// envconfig() selects [node_db] type=rwdb, which is the tree-only null node
+// store: SHAMapStore never builds DatabasePinned on it, so catalogue_load
+// would refuse to run. Use the in-memory "memory" store for the hot side
+// instead. It is shared by path within the process, so each config gets its
+// own path (SHAMapStore creates the directory; nothing is written to it).
+// The pinned side uses rwdb, an in-memory persistent backend.
 inline std::unique_ptr<Config>
 catalogueEnvconfig(CatalogueEnvOpts const& opts = {})
 {
     auto cfg = test::jtx::envconfig();
     auto& nodeDb = cfg->section(ConfigSection::nodeDatabase());
     if (opts.withPinnedType)
+    {
+        nodeDb.set("type", "memory");
+        nodeDb.set(
+            "path",
+            (boost::filesystem::temp_directory_path() /
+             boost::filesystem::unique_path("catalogue-test-%%%%-%%%%-%%%%"))
+                .string());
         nodeDb.set("pinned_type", "rwdb");
+    }
     nodeDb.set("online_delete", opts.onlineDelete);
     return cfg;
 }
@@ -1067,6 +1087,621 @@ class Catalogue_test : public beast::unit_test::suite
     }
 
 public:
+    // ---- Editing uncompressed catalogues -----------------------------------
+    //
+    // Ledger record: seq, hash, txHash, accountHash, parentHash, drops,
+    // closeFlags, closeTimeResolution, closeTime, parentCloseTime; then the
+    // state map and the tx map, each a run of leaf records ending in
+    // tnTERMINAL. A leaf record is type (1), key (32), and, unless it is a
+    // tnREMOVE, size (4) and data.
+
+    static constexpr std::size_t catLedgerHeaderSize =
+        4 + 32 * 4 + 8 + sizeof(int) + 4 + 8 + 8;
+
+    struct CatLedgerRecord
+    {
+        std::size_t header;  // first byte of the ledger header
+        std::size_t state;   // first byte of the state map
+        std::size_t tx;      // first byte of the tx map
+        std::size_t end;     // one past the tx map's tnTERMINAL
+    };
+
+    // Position just past the map that starts at `pos`.
+    static std::size_t
+    skipCatalogueMap(std::vector<std::uint8_t> const& bytes, std::size_t pos)
+    {
+        while (pos < bytes.size())
+        {
+            auto const type = bytes[pos++];
+            if (type == 255)  // tnTERMINAL
+                return pos;
+            pos += 32;
+            if (type == 254)  // tnREMOVE
+                continue;
+            if (pos + 4 > bytes.size())
+                break;
+            std::uint32_t size = 0;
+            std::memcpy(&size, &bytes[pos], 4);
+            pos += 4 + size;
+        }
+        return bytes.size();
+    }
+
+    // Position of the size field of the first leaf (not a removal) in the
+    // map that starts at `pos`, if the map has one.
+    static std::optional<std::size_t>
+    firstLeafSizeField(std::vector<std::uint8_t> const& bytes, std::size_t pos)
+    {
+        while (pos < bytes.size())
+        {
+            auto const type = bytes[pos];
+            if (type == 255)
+                return std::nullopt;
+            if (type != 254)
+                return pos + 1 + 32;
+            pos += 1 + 32;
+        }
+        return std::nullopt;
+    }
+
+    static std::vector<CatLedgerRecord>
+    indexCatalogue(std::vector<std::uint8_t> const& bytes)
+    {
+        std::vector<CatLedgerRecord> records;
+        std::size_t pos = sizeof(TestCATLHeader);
+        while (pos + catLedgerHeaderSize <= bytes.size())
+        {
+            CatLedgerRecord r;
+            r.header = pos;
+            r.state = pos + catLedgerHeaderSize;
+            r.tx = skipCatalogueMap(bytes, r.state);
+            r.end = skipCatalogueMap(bytes, r.tx);
+            records.push_back(r);
+            pos = r.end;
+        }
+        return records;
+    }
+
+    static std::vector<std::uint8_t>
+    readCatalogue(std::string const& path)
+    {
+        std::ifstream in(path, std::ios::binary);
+        return {
+            std::istreambuf_iterator<char>(in),
+            std::istreambuf_iterator<char>()};
+    }
+
+    static void
+    writeCatalogue(std::string const& path, std::vector<std::uint8_t> const& b)
+    {
+        std::ofstream out(path, std::ios::binary | std::ios::trunc);
+        out.write(reinterpret_cast<char const*>(b.data()), b.size());
+    }
+
+    static Json::Value
+    loadCatalogue(test::jtx::Env& env, std::string const& path, bool ignoreHash)
+    {
+        Json::Value params{Json::objectValue};
+        params[jss::input_file] = path;
+        if (ignoreHash)
+            params[jss::ignore_hash] = true;
+        return env.client().invoke("catalogue_load", params)[jss::result];
+    }
+
+    static bool
+    errorContains(Json::Value const& result, std::string const& what)
+    {
+        return result[jss::status] == "error" &&
+            result[jss::error_message].asString().find(what) !=
+            std::string::npos;
+    }
+
+    void
+    testCatalogueLoadFailures(FeatureBitset features)
+    {
+        testcase("catalogue_load: failures part way through a file");
+        using namespace test::jtx;
+        using beast::severities::kDisabled;
+
+        boost::filesystem::path tempDir =
+            boost::filesystem::temp_directory_path() /
+            boost::filesystem::unique_path();
+        boost::filesystem::create_directories(tempDir);
+        auto const goodPath = (tempDir / "good.catl").string();
+
+        // An uncompressed catalogue of ledgers 3-6, so records can be edited.
+        {
+            Env env{*this, catalogueEnvconfig(), features, nullptr, kDisabled};
+            prepareLedgerData(env, 3);
+
+            Json::Value params{Json::objectValue};
+            params[jss::min_ledger] = 3;
+            params[jss::max_ledger] = 6;
+            params[jss::output_file] = goodPath;
+            params[jss::compression_level] = 0;
+            auto const result =
+                env.client().invoke("catalogue_create", params)[jss::result];
+            BEAST_EXPECT(result[jss::status] == jss::success);
+        }
+
+        auto const good = readCatalogue(goodPath);
+        auto const records = indexCatalogue(good);
+        if (!BEAST_EXPECT(
+                records.size() == 4 && records.back().end == good.size()))
+            return;
+
+        // Write `bytes` to a new file, load it into a fresh server, and hand
+        // the result and the resulting pinned set to `check`.
+        int fileNo = 0;
+        auto loadEdited = [&](std::vector<std::uint8_t> const& bytes,
+                              bool ignoreHash,
+                              auto&& check) {
+            auto const path =
+                (tempDir / ("edited" + std::to_string(++fileNo) + ".catl"))
+                    .string();
+            writeCatalogue(path, bytes);
+            Env env{*this, catalogueEnvconfig(), features, nullptr, kDisabled};
+            auto const result = loadCatalogue(env, path, ignoreHash);
+            check(
+                result, env.app().getLedgerMaster().getPinnedLedgersRangeSet());
+        };
+
+        auto const hashField = offsetof(TestCATLHeader, hash);
+        auto const maxLedgerField = offsetof(TestCATLHeader, max_ledger);
+        // Offsets within a ledger header record.
+        constexpr std::size_t accountHashAt = 4 + 32 * 2;
+        constexpr std::size_t dropsAt = 4 + 32 * 4;
+
+        // The stored checksum is wrong but every ledger is intact: all four
+        // load, then the whole-file check fails and the run is rolled back.
+        {
+            auto bytes = good;
+            bytes[hashField] ^= 0xff;
+            loadEdited(
+                bytes, false, [&](auto const& result, auto const& pinned) {
+                    BEAST_EXPECT(errorContains(
+                        result, "catalogue hash verification failed"));
+                    BEAST_EXPECT(pinned.empty());
+                });
+        }
+
+        // A ledger part way through is corrupt. The load stops there, the
+        // rest of the file is still hashed, the checksum fails, and the
+        // ledgers loaded before it are rolled back too.
+        {
+            auto bytes = good;
+            bytes[records[2].header + dropsAt] ^= 0x01;
+            loadEdited(
+                bytes, false, [&](auto const& result, auto const& pinned) {
+                    BEAST_EXPECT(errorContains(
+                        result, "catalogue hash verification failed"));
+                    BEAST_EXPECT(pinned.empty());
+                });
+        }
+
+        // A checksum failure only rolls back what this run added: ledgers
+        // that were already pinned stay pinned.
+        {
+            auto bytes = good;
+            bytes[hashField] ^= 0xff;
+            auto const badPath = (tempDir / "badsum.catl").string();
+            writeCatalogue(badPath, bytes);
+
+            Env env{*this, catalogueEnvconfig(), features, nullptr, kDisabled};
+            BEAST_EXPECT(
+                loadCatalogue(env, goodPath, false)[jss::status] ==
+                jss::success);
+            BEAST_EXPECT(errorContains(
+                loadCatalogue(env, badPath, false),
+                "catalogue hash verification failed"));
+            auto& lm = env.app().getLedgerMaster();
+            BEAST_EXPECT(to_string(lm.getPinnedLedgersRangeSet()) == "3-6");
+
+            // The pinned store: an empty batch is a no-op, an in-memory
+            // backend is durable on demand, and a fetch made for
+            // online_delete's copy pass (duplicate == true) finds nodes
+            // that exist only in the pinned store.
+            auto* const pinnedDb = dynamic_cast<NodeStore::DatabasePinnedImp*>(
+                &env.app().getNodeStore());
+            if (BEAST_EXPECT(pinnedDb))
+            {
+                pinnedDb->storePinnedBatch({});
+                BEAST_EXPECT(pinnedDb->syncPinned());
+                auto const stateRoot =
+                    uint256::fromVoid(&good[records[0].header + accountHashAt]);
+                BEAST_EXPECT(env.app().getNodeStore().fetchNodeObject(
+                    stateRoot, 0, NodeStore::FetchType::synchronous, true));
+            }
+        }
+
+        // Out of sequence: ledger 4's record claims to be ledger 9.
+        {
+            auto bytes = good;
+            std::uint32_t const nine = 9;
+            std::memcpy(&bytes[records[1].header], &nine, 4);
+            loadEdited(
+                bytes, true, [&](auto const& result, auto const& pinned) {
+                    BEAST_EXPECT(errorContains(result, "out of sequence"));
+                    BEAST_EXPECT(boost::icl::contains(pinned, 3u));
+                    BEAST_EXPECT(!boost::icl::contains(pinned, 4u));
+                });
+        }
+
+        // The header promises a ledger 7 the file does not contain. Ledgers
+        // 3-6 load (hash checks are off) before the file runs out.
+        {
+            auto bytes = good;
+            std::uint32_t const seven = 7;
+            std::memcpy(&bytes[maxLedgerField], &seven, 4);
+            loadEdited(
+                bytes, true, [&](auto const& result, auto const& pinned) {
+                    BEAST_EXPECT(errorContains(
+                        result, "Unexpected end of catalogue file"));
+                    BEAST_EXPECT(to_string(pinned) == "3-6");
+                });
+        }
+
+        // An absurd leaf size in each kind of map is rejected while parsing.
+        auto withHugeLeaf = [&](std::size_t sizeField) {
+            auto bytes = good;
+            std::uint32_t const huge = 0xFFFFFFFF;
+            std::memcpy(&bytes[sizeField], &huge, 4);
+            return bytes;
+        };
+
+        // ... in the base ledger's state map.
+        if (auto const field = firstLeafSizeField(good, records[0].state);
+            BEAST_EXPECT(field))
+        {
+            loadEdited(
+                withHugeLeaf(*field),
+                true,
+                [&](auto const& result, auto const& pinned) {
+                    BEAST_EXPECT(errorContains(
+                        result, "Failed to load base ledger state"));
+                    BEAST_EXPECT(pinned.empty());
+                });
+        }
+
+        // ... in a later ledger's state delta.
+        {
+            std::optional<std::size_t> index;
+            std::optional<std::size_t> field;
+            for (std::size_t i = 1; i < records.size() && !field; ++i)
+                if ((field = firstLeafSizeField(good, records[i].state)))
+                    index = i;
+            if (BEAST_EXPECT(field))
+            {
+                std::uint32_t const seq = 3 + *index;
+                loadEdited(
+                    withHugeLeaf(*field),
+                    true,
+                    [&](auto const& result, auto const& pinned) {
+                        BEAST_EXPECT(errorContains(
+                            result, "Failed to apply ledger delta"));
+                        BEAST_EXPECT(boost::icl::contains(pinned, seq - 1));
+                        BEAST_EXPECT(!boost::icl::contains(pinned, seq));
+                    });
+            }
+        }
+
+        // ... in a ledger's transaction map.
+        {
+            std::optional<std::size_t> index;
+            std::optional<std::size_t> field;
+            for (std::size_t i = 0; i < records.size() && !field; ++i)
+                if ((field = firstLeafSizeField(good, records[i].tx)))
+                    index = i;
+            if (BEAST_EXPECT(field))
+            {
+                std::uint32_t const seq = 3 + *index;
+                loadEdited(
+                    withHugeLeaf(*field),
+                    true,
+                    [&](auto const& result, auto const& pinned) {
+                        BEAST_EXPECT(errorContains(
+                            result, "Failed to apply ledger delta"));
+                        BEAST_EXPECT(!boost::icl::contains(pinned, seq));
+                    });
+            }
+        }
+
+        boost::filesystem::remove_all(tempDir);
+    }
+
+    void
+    testCataloguePinnedStoreRequired(FeatureBitset features)
+    {
+        testcase("catalogue_load / catalogue_unpin: pinned store required");
+        using namespace test::jtx;
+        using beast::severities::kDisabled;
+
+        // pinned_type is set, but [node_db] type=rwdb is the null node
+        // store, on which SHAMapStore never builds the pinned store.
+        {
+            auto cfg = test::jtx::envconfig();
+            auto& nodeDb = cfg->section(ConfigSection::nodeDatabase());
+            nodeDb.set("pinned_type", "rwdb");
+            nodeDb.set("online_delete", "256");
+            Env env{*this, std::move(cfg), features, nullptr, kDisabled};
+
+            Json::Value params{Json::objectValue};
+            params[jss::input_file] = "/nonexistent.catl";
+            auto const result =
+                env.client().invoke("catalogue_load", params)[jss::result];
+            BEAST_EXPECT(
+                errorContains(result, "requires the pinned node store"));
+        }
+
+        // catalogue_unpin needs pinned_type configured at all.
+        {
+            Env env{
+                *this,
+                catalogueEnvconfig({.withPinnedType = false}),
+                features,
+                nullptr,
+                kDisabled};
+
+            Json::Value params{Json::objectValue};
+            params[jss::ledger_index_min] = 1;
+            params[jss::ledger_index_max] = 2;
+            auto const result =
+                env.client().invoke("catalogue_unpin", params)[jss::result];
+            BEAST_EXPECT(errorContains(result, "pinned_type"));
+        }
+    }
+
+    void
+    testCatalogueChainLinkage(FeatureBitset features)
+    {
+        testcase("catalogue_load: Ledgers must chain");
+        using namespace test::jtx;
+
+        Env env{
+            *this,
+            catalogueEnvconfig(),
+            features,
+            nullptr,
+            beast::severities::kDisabled,
+        };
+        prepareLedgerData(env, 3);
+
+        boost::filesystem::path tempDir =
+            boost::filesystem::temp_directory_path() /
+            boost::filesystem::unique_path();
+        boost::filesystem::create_directories(tempDir);
+        auto cataloguePath = (tempDir / "test.catl").string();
+
+        // Uncompressed, so ledger records can be located and edited.
+        {
+            Json::Value params{Json::objectValue};
+            params[jss::min_ledger] = 3;
+            params[jss::max_ledger] = 6;
+            params[jss::output_file] = cataloguePath;
+            params[jss::compression_level] = 0;
+            auto const result =
+                env.client().invoke("catalogue_create", params)[jss::result];
+            BEAST_EXPECT(result[jss::status] == jss::success);
+        }
+
+        std::vector<std::uint8_t> bytes;
+        {
+            std::ifstream in(cataloguePath, std::ios::binary);
+            bytes.assign(
+                std::istreambuf_iterator<char>(in),
+                std::istreambuf_iterator<char>());
+        }
+
+        // Ledger record: seq, hash, txHash, accountHash, parentHash, drops,
+        // closeFlags, closeTimeResolution, closeTime, parentCloseTime; then
+        // the state map and tx map as leaf records ending in tnTERMINAL.
+        constexpr std::size_t ledgerHeaderSize =
+            4 + 32 * 4 + 8 + sizeof(int) + 4 + 8 + 8;
+        auto skipMap = [&](std::size_t pos) -> std::size_t {
+            while (pos < bytes.size())
+            {
+                auto const type = bytes[pos++];
+                if (type == 255)  // tnTERMINAL
+                    return pos;
+                pos += 32;
+                if (type == 254)  // tnREMOVE
+                    continue;
+                if (pos + 4 > bytes.size())
+                    break;
+                std::uint32_t size = 0;
+                std::memcpy(&size, &bytes[pos], 4);
+                pos += 4 + size;
+            }
+            return bytes.size();
+        };
+
+        // Skip the base ledger (3) to reach ledger 4.
+        std::size_t pos = sizeof(TestCATLHeader) + ledgerHeaderSize;
+        pos = skipMap(skipMap(pos));
+        if (!BEAST_EXPECT(pos + ledgerHeaderSize <= bytes.size()))
+            return;
+
+        LedgerInfo info;
+        std::size_t p = pos;
+        auto take = [&](void* dst, std::size_t n) {
+            std::memcpy(dst, &bytes[p], n);
+            p += n;
+        };
+        std::uint64_t drops = 0, closeTime = 0, parentCloseTime = 0;
+        std::uint32_t resolution = 0;
+        take(&info.seq, 4);
+        std::size_t const hashOffset = p;
+        take(info.hash.data(), 32);
+        take(info.txHash.data(), 32);
+        take(info.accountHash.data(), 32);
+        std::size_t const parentOffset = p;
+        take(info.parentHash.data(), 32);
+        take(&drops, 8);
+        take(&info.closeFlags, sizeof(info.closeFlags));
+        take(&resolution, 4);
+        take(&closeTime, 8);
+        take(&parentCloseTime, 8);
+        info.drops = drops;
+        info.closeTime = NetClock::time_point{NetClock::duration{closeTime}};
+        info.parentCloseTime =
+            NetClock::time_point{NetClock::duration{parentCloseTime}};
+        info.closeTimeResolution = NetClock::duration{resolution};
+
+        BEAST_EXPECT(info.seq == 4);
+        // Sanity: the record parses to a ledger whose hash checks out.
+        BEAST_EXPECT(calculateLedgerHash(info) == info.hash);
+
+        // Re-parent ledger 4 and fix up its hash, so the file stays
+        // internally consistent ledger by ledger; only the chain check can
+        // catch it.
+        info.parentHash = ~info.parentHash;
+        info.hash = calculateLedgerHash(info);
+        std::memcpy(&bytes[parentOffset], info.parentHash.data(), 32);
+        std::memcpy(&bytes[hashOffset], info.hash.data(), 32);
+        {
+            std::ofstream out(
+                cataloguePath, std::ios::binary | std::ios::trunc);
+            out.write(
+                reinterpret_cast<char const*>(bytes.data()), bytes.size());
+        }
+
+        Env loadEnv{
+            *this,
+            catalogueEnvconfig(),
+            features,
+            nullptr,
+            beast::severities::kDisabled,
+        };
+
+        Json::Value params{Json::objectValue};
+        params[jss::input_file] = cataloguePath;
+        params[jss::ignore_hash] = true;  // the file was edited
+        auto const result =
+            loadEnv.client().invoke("catalogue_load", params)[jss::result];
+        BEAST_EXPECT(result[jss::status] == "error");
+        BEAST_EXPECT(
+            result[jss::error_message].asString().find(
+                "does not chain to its predecessor") != std::string::npos);
+
+        // The base ledger verified and stays pinned; nothing after it does.
+        auto const pinned =
+            loadEnv.app().getLedgerMaster().getPinnedLedgersRangeSet();
+        BEAST_EXPECT(boost::icl::contains(pinned, 3u));
+        BEAST_EXPECT(!boost::icl::contains(pinned, 4u));
+
+        boost::filesystem::remove_all(tempDir);
+    }
+
+    void
+    testCatalogueUnpin(FeatureBitset features)
+    {
+        testcase("catalogue_unpin");
+        using namespace test::jtx;
+
+        Env env{
+            *this,
+            catalogueEnvconfig(),
+            features,
+            nullptr,
+            beast::severities::kDisabled,
+        };
+        prepareLedgerData(env, 5);
+
+        boost::filesystem::path tempDir =
+            boost::filesystem::temp_directory_path() /
+            boost::filesystem::unique_path();
+        boost::filesystem::create_directories(tempDir);
+        auto cataloguePath = (tempDir / "test.catl").string();
+
+        {
+            Json::Value params{Json::objectValue};
+            params[jss::min_ledger] = 3;
+            params[jss::max_ledger] = 8;
+            params[jss::output_file] = cataloguePath;
+            auto const result =
+                env.client().invoke("catalogue_create", params)[jss::result];
+            BEAST_EXPECT(result[jss::status] == jss::success);
+        }
+        {
+            Json::Value params{Json::objectValue};
+            params[jss::input_file] = cataloguePath;
+            auto const result =
+                env.client().invoke("catalogue_load", params)[jss::result];
+            BEAST_EXPECT(result[jss::status] == jss::success);
+        }
+
+        auto& lm = env.app().getLedgerMaster();
+        BEAST_EXPECT(to_string(lm.getPinnedLedgersRangeSet()) == "3-8");
+
+        auto unpin = [&](std::optional<Json::Value> min,
+                         std::optional<Json::Value> max) {
+            Json::Value params{Json::objectValue};
+            if (min)
+                params[jss::ledger_index_min] = *min;
+            if (max)
+                params[jss::ledger_index_max] = *max;
+            return env.client().invoke("catalogue_unpin", params)[jss::result];
+        };
+
+        // Bad parameters
+        BEAST_EXPECT(unpin({}, {})[jss::status] == "error");
+        BEAST_EXPECT(unpin(5, {})[jss::status] == "error");
+        BEAST_EXPECT(unpin(6, 5)[jss::status] == "error");
+        BEAST_EXPECT(unpin(-1, 5)[jss::status] == "error");
+        BEAST_EXPECT(unpin("x", 5)[jss::status] == "error");
+        BEAST_EXPECT(to_string(lm.getPinnedLedgersRangeSet()) == "3-8");
+
+        // Cut a hole
+        {
+            auto const result = unpin(5, 6);
+            BEAST_EXPECT(result[jss::status] == jss::success);
+            BEAST_EXPECT(result[jss::unpinned] == "5-6");
+            BEAST_EXPECT(result[jss::complete_ledgers_pinned] == "3-4,7-8");
+            BEAST_EXPECT(lm.isPinned(4) && lm.isPinned(7));
+            BEAST_EXPECT(!lm.isPinned(5) && !lm.isPinned(6));
+        }
+
+        // A range only partly pinned unpins just the pinned part
+        {
+            auto const result = unpin(1, 4);
+            BEAST_EXPECT(result[jss::status] == jss::success);
+            BEAST_EXPECT(result[jss::unpinned] == "3-4");
+            BEAST_EXPECT(result[jss::complete_ledgers_pinned] == "7-8");
+        }
+
+        // Nothing pinned in range: success, nothing unpinned. (Unsigned
+        // JSON values are accepted as well as non-negative signed ones.)
+        {
+            auto const result = unpin(Json::UInt(5), Json::UInt(6));
+            BEAST_EXPECT(result[jss::status] == jss::success);
+            BEAST_EXPECT(result[jss::unpinned] == "empty");
+            BEAST_EXPECT(result[jss::complete_ledgers_pinned] == "7-8");
+        }
+
+        // Unpinning the rest leaves nothing pinned. (state.db then records
+        // an empty set, which must not stop the next start; see
+        // SHAMapStore_test.)
+        {
+            auto const result = unpin(1, 100);
+            BEAST_EXPECT(result[jss::status] == jss::success);
+            BEAST_EXPECT(result[jss::unpinned] == "7-8");
+            BEAST_EXPECT(result[jss::complete_ledgers_pinned] == "empty");
+            BEAST_EXPECT(lm.getPinnedLedgersRangeSet().empty());
+        }
+
+        // Re-loading the file pins it again
+        {
+            Json::Value params{Json::objectValue};
+            params[jss::input_file] = cataloguePath;
+            auto const result =
+                env.client().invoke("catalogue_load", params)[jss::result];
+            BEAST_EXPECT(result[jss::status] == jss::success);
+            BEAST_EXPECT(to_string(lm.getPinnedLedgersRangeSet()) == "3-8");
+        }
+
+        boost::filesystem::remove_all(tempDir);
+    }
+
     void
     run() override
     {
@@ -1081,6 +1716,10 @@ public:
         testCatalogueFileSize(all);
         testCatalogueCompression(all);
         testCatalogueStatus(all);
+        testCatalogueChainLinkage(all);
+        testCatalogueUnpin(all);
+        testCatalogueLoadFailures(all);
+        testCataloguePinnedStoreRequired(all);
     }
 };
 

src/test/rpc/RPCCall_test.cpp

@@ -1934,6 +1934,56 @@ static RPCCallTestData const rpcCallTestArray[] = {
      RPCCallTestData::bad_cast,
      R"()"},
 
+    // catalogue_unpin
+    // ------------------------------------------------------------
+    {"catalogue_unpin: minimal.",
+     __LINE__,
+     {"catalogue_unpin", "5000001", "6000000"},
+     RPCCallTestData::no_exception,
+     R"({
+    "method" : "catalogue_unpin",
+    "params" : [
+      {
+         "api_version" : %API_VER%,
+         "ledger_index_max" : 6000000,
+         "ledger_index_min" : 5000001
+      }
+    ]
+    })"},
+    {"catalogue_unpin: too few arguments.",
+     __LINE__,
+     {"catalogue_unpin", "5000001"},
+     RPCCallTestData::no_exception,
+     R"({
+    "method" : "catalogue_unpin",
+    "params" : [
+      {
+         "error" : "badSyntax",
+         "error_code" : 1,
+         "error_message" : "Syntax error."
+      }
+    ]
+    })"},
+    {"catalogue_unpin: too many arguments.",
+     __LINE__,
+     {"catalogue_unpin", "1", "2", "3"},
+     RPCCallTestData::no_exception,
+     R"({
+    "method" : "catalogue_unpin",
+    "params" : [
+      {
+         "error" : "badSyntax",
+         "error_code" : 1,
+         "error_message" : "Syntax error."
+      }
+    ]
+    })"},
+    {"catalogue_unpin: non-integer bound.",
+     __LINE__,
+     {"catalogue_unpin", "first", "6000000"},
+     RPCCallTestData::bad_cast,
+     R"()"},
+
     // channel_authorize
     // -----------------------------------------------------------
     {"channel_authorize: minimal.",

src/test/shamap/SHAMap_test.cpp

@@ -29,7 +29,9 @@
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/protocol/Serializer.h>
 
+#include <algorithm>
 #include <optional>
+#include <vector>
 
 namespace ripple {
 namespace tests {
@@ -131,6 +133,7 @@ public:
         testCanonicalizeMergesInnerChildren(journal);
         testCanonicalizeSharesDuplicateContent(journal);
         testGetLinkedNode(journal);
+        testUpsertAndFlushToSink(journal);
     }
 
     void
@@ -284,6 +287,82 @@ public:
         BEAST_EXPECT(liveLinked > 0);
     }
 
+    void
+    testUpsertAndFlushToSink(beast::Journal const& journal)
+    {
+        testcase("upsertGiveItem and flushDirty to a sink");
+
+        tests::TestNodeFamily f(journal);
+        SHAMap a{SHAMapType::FREE, f};  // built with upsertGiveItem
+        SHAMap b{SHAMapType::FREE, f};  // reference: addItem/updateGiveItem
+
+        // Distinct first nibbles: a root with three leaf children.
+        constexpr uint256 h1(
+            "092891fe4ef6cee585fdc6fda0e09eb4d386363158ec3321b8123e5a772c6ca7");
+        constexpr uint256 h2(
+            "436ccbac3347baa1f1e53baeef1f43334da88f1f6d70d963b833afd6dfa289fe");
+        constexpr uint256 h3(
+            "b92891fe4ef6cee585fdc6fda1e09eb4d386363158ec3321b8123e5a772c6ca8");
+        auto const nm = SHAMapNodeType::tnTRANSACTION_NM;
+
+        // Absent keys are added.
+        BEAST_EXPECT(a.upsertGiveItem(nm, make_shamapitem(h1, IntToVUC(1))));
+        BEAST_EXPECT(a.upsertGiveItem(nm, make_shamapitem(h2, IntToVUC(2))));
+        BEAST_EXPECT(a.upsertGiveItem(nm, make_shamapitem(h3, IntToVUC(3))));
+        // A present key has its data replaced.
+        BEAST_EXPECT(a.upsertGiveItem(nm, make_shamapitem(h2, IntToVUC(20))));
+        // A change of leaf type is refused and leaves the item untouched.
+        BEAST_EXPECT(!a.upsertGiveItem(
+            SHAMapNodeType::tnTRANSACTION_MD,
+            make_shamapitem(h1, IntToVUC(10))));
+
+        BEAST_EXPECT(b.addItem(nm, make_shamapitem(h1, IntToVUC(1))));
+        BEAST_EXPECT(b.addItem(nm, make_shamapitem(h2, IntToVUC(2))));
+        BEAST_EXPECT(b.addItem(nm, make_shamapitem(h3, IntToVUC(3))));
+        BEAST_EXPECT(b.updateGiveItem(nm, make_shamapitem(h2, IntToVUC(20))));
+
+        {
+            auto const v20 = IntToVUC(20);
+            auto const& item = a.peekItem(h2);
+            BEAST_EXPECT(
+                item && item->slice() == Slice(v20.data(), v20.size()));
+            auto const v1 = IntToVUC(1);
+            auto const& item1 = a.peekItem(h1);
+            BEAST_EXPECT(
+                item1 && item1->slice() == Slice(v1.data(), v1.size()));
+        }
+
+        // Flush to a sink. Do not call getHash() on `a` first: it unshares
+        // the map and the flush would then have nothing to write.
+        auto const cacheBefore = f.getTreeNodeCache()->size();
+        std::vector<uint256> written;
+        bool sinkArgsOk = true;
+        SHAMap::FlushSink const sink =
+            [&](NodeObjectType t, Blob&& data, uint256 const& hash) {
+                sinkArgsOk =
+                    sinkArgsOk && t == hotTRANSACTION_NODE && !data.empty();
+                written.push_back(hash);
+            };
+        a.flushDirty(hotTRANSACTION_NODE, sink);
+
+        BEAST_EXPECT(sinkArgsOk);
+        BEAST_EXPECT(written.size() == 4);  // root + three leaves
+        // Unlike flushDirty(t), nothing was canonicalized into the cache.
+        BEAST_EXPECT(f.getTreeNodeCache()->size() == cacheBefore);
+
+        // The upserted map matches the reference, and its root was written.
+        BEAST_EXPECT(a.getHash() == b.getHash());
+        BEAST_EXPECT(
+            std::find(
+                written.begin(), written.end(), a.getHash().as_uint256()) !=
+            written.end());
+
+        // Everything is now shared: a second flush writes nothing.
+        written.clear();
+        a.flushDirty(hotTRANSACTION_NODE, sink);
+        BEAST_EXPECT(written.empty());
+    }
+
     void
     testGetLinkedNode(beast::Journal const& journal)
     {

src/xrpld/app/ledger/AcceptedLedger.cpp

@@ -25,7 +25,8 @@ namespace ripple {
 
 AcceptedLedger::AcceptedLedger(
     std::shared_ptr<ReadView const> const& ledger,
-    Application& app)
+    Application& app,
+    bool withJson)
     : mLedger(ledger)
 {
     transactions_.reserve(256);
@@ -33,7 +34,7 @@ AcceptedLedger::AcceptedLedger(
     auto insertAll = [&](auto const& txns) {
         for (auto const& item : txns)
             transactions_.emplace_back(std::make_unique<AcceptedLedgerTx>(
-                ledger, item.first, item.second));
+                ledger, item.first, item.second, withJson));
     };
 
     transactions_.reserve(256);

src/xrpld/app/ledger/AcceptedLedger.h

@@ -44,9 +44,15 @@ namespace ripple {
 class AcceptedLedger : public CountedObject<AcceptedLedger>
 {
 public:
+    /** @param withJson build each transaction's publishable JSON. Pass
+               false only for ledgers that will never be published from this
+               object (e.g. pinned catalogue history being saved to SQL);
+               getJson() on such transactions returns a null value.
+    */
     AcceptedLedger(
         std::shared_ptr<ReadView const> const& ledger,
-        Application& app);
+        Application& app,
+        bool withJson = true);
 
     std::shared_ptr<ReadView const> const&
     getLedger() const

src/xrpld/app/ledger/AcceptedLedgerTx.cpp

@@ -28,7 +28,8 @@ namespace ripple {
 AcceptedLedgerTx::AcceptedLedgerTx(
     std::shared_ptr<ReadView const> const& ledger,
     std::shared_ptr<STTx const> const& txn,
-    std::shared_ptr<STObject const> const& met)
+    std::shared_ptr<STObject const> const& met,
+    bool withJson)
     : mTxn(txn)
     , mMeta(txn->getTransactionID(), ledger->seq(), *met)
     , mAffected(mMeta.getAffectedAccounts())
@@ -41,6 +42,11 @@ AcceptedLedgerTx::AcceptedLedgerTx(
     met->add(s);
     mRawMeta = std::move(s.modData());
 
+    // The JSON (full tx + meta + hex raw meta per transaction) is only
+    // needed for publishing. Bulk SQL saves of pinned history skip it.
+    if (!withJson)
+        return;
+
     mJson = Json::objectValue;
     mJson[jss::transaction] = mTxn->getJson(JsonOptions::none);
 

src/xrpld/app/ledger/AcceptedLedgerTx.h

@@ -46,7 +46,8 @@ public:
     AcceptedLedgerTx(
         std::shared_ptr<ReadView const> const& ledger,
         std::shared_ptr<STTx const> const&,
-        std::shared_ptr<STObject const> const&);
+        std::shared_ptr<STObject const> const&,
+        bool withJson = true);
 
     std::shared_ptr<STTx const> const&
     getTxn() const
@@ -88,6 +89,7 @@ public:
     std::string
     getEscMeta() const;
 
+    /** Null if constructed with withJson == false. */
     Json::Value const&
     getJson() const
     {

src/xrpld/app/ledger/LedgerMaster.h

@@ -228,6 +228,19 @@ public:
     isPinned(std::uint32_t seq);
     void
     unpinLedger(std::uint32_t seq);
+
+    /** Remove `ranges` from the pinned set.
+
+        @param dropFromComplete of the ledgers actually unpinned, those to
+               also remove from complete_ledgers now (used to roll back a
+               load). Otherwise unpinned ledgers stay complete until online
+               delete releases them.
+        @return the ledgers that were pinned and no longer are.
+    */
+    RangeSet<std::uint32_t>
+    unpinLedgers(
+        RangeSet<std::uint32_t> const& ranges,
+        RangeSet<std::uint32_t> const& dropFromComplete = {});
     bool
     isValidated(ReadView const& ledger);
     bool

src/xrpld/app/ledger/detail/LedgerMaster.cpp

@@ -501,7 +501,9 @@ LedgerMaster::storeLedger(std::shared_ptr<Ledger const> ledger, bool pin)
             std::lock_guard sl(mPinnedLock);
             mPinnedLedgers.insert(range(seq, seq));
         }
-        JLOG(m_journal.info()) << "Pinned ledger : " << seq;
+        // catalogue_load pins every ledger it imports; at info this would
+        // emit one line per ledger for the whole file.
+        JLOG(m_journal.debug()) << "Pinned ledger : " << seq;
     }
     return true;
 }
@@ -586,6 +588,19 @@ LedgerMaster::unpinLedger(std::uint32_t seq)
     mPinnedLedgers.erase(range(seq, seq));
 }
 
+RangeSet<std::uint32_t>
+LedgerMaster::unpinLedgers(
+    RangeSet<std::uint32_t> const& ranges,
+    RangeSet<std::uint32_t> const& dropFromComplete)
+{
+    std::scoped_lock lock(mCompleteLock, mPinnedLock);
+    RangeSet<std::uint32_t> removed = mPinnedLedgers & ranges;
+    mPinnedLedgers -= removed;
+    if (!dropFromComplete.empty())
+        mCompleteLedgers -= (removed & dropFromComplete);
+    return removed;
+}
+
 bool
 LedgerMaster::isValidated(ReadView const& ledger)
 {

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -381,7 +381,9 @@ void
 SHAMapStoreImp::loadPinnedRanges()
 {
     auto rangesStr = state_db_.getPinnedRanges();
-    if (rangesStr.empty())
+    // "empty" is what to_string() renders for an empty set; accept it in
+    // case it was stored before setPinnedRanges() learned to write "".
+    if (rangesStr.empty() || rangesStr == "empty")
         return;
 
     auto const& nscfg = app_.config().section(ConfigSection::nodeDatabase());
@@ -411,22 +413,62 @@ SHAMapStoreImp::loadPinnedRanges()
             static_cast<bool>(app_.getNodeStore().fetchNodeObject(hash, seq));
     };
 
+    //
+    // catalogue_load records ranges at checkpoints, after handing the pinned
+    // backend every node of the recorded ledgers, but not every backend can
+    // make those writes durable on demand (NuDB commits on its own thread).
+    // A crash can therefore leave the tail of an interval recorded but not
+    // on disk. Each ledger's nodes reach the backend before its header, and
+    // ledgers in sequence, so what survives is a prefix: trim the upper bound
+    // back to the last ledger whose header is present instead of refusing to
+    // start. A missing lower bound still means the configured pinned store
+    // is not the one these ranges were loaded into.
+    constexpr std::uint32_t maxTailTrim = 65536;
+
+    RangeSet<std::uint32_t> verified;
     for (auto const& interval : persistedRanges)
     {
-        if (!hasLedgerData(interval.lower()) ||
-            !hasLedgerData(interval.upper()))
-        {
-            Throw<std::runtime_error>(
-                "Persisted pinned interval " +
-                std::to_string(interval.lower()) + "-" +
+        auto const lower = interval.lower();
+        auto upper = interval.upper();
+
+        auto const notPresent = [&]() {
+            return std::runtime_error(
+                "Persisted pinned interval " + std::to_string(lower) + "-" +
                 std::to_string(interval.upper()) +
                 " is not present in the currently configured pinned store");
+        };
+
+        if (!hasLedgerData(lower))
+            Throw<std::runtime_error>(notPresent());
+
+        std::uint32_t trimmed = 0;
+        while (upper > lower && !hasLedgerData(upper))
+        {
+            if (++trimmed > maxTailTrim)
+                Throw<std::runtime_error>(notPresent());
+            --upper;
         }
+
+        if (trimmed)
+        {
+            JLOG(journal_.warn())
+                << "Pinned interval " << lower << "-" << interval.upper()
+                << " was recorded but its last " << trimmed
+                << " ledger(s) are not in the pinned store (likely an "
+                   "interrupted catalogue_load); trimmed to "
+                << lower << "-" << upper
+                << ". Re-run catalogue_load to restore them.";
+        }
+
+        verified.insert(range(lower, upper));
     }
 
+    if (verified != persistedRanges)
+        state_db_.setPinnedRanges(to_string(verified));
+
     JLOG(journal_.info()) << "Loaded pinned ranges from database: "
-                          << rangesStr;
-    app_.getLedgerMaster().setPinnedLedgersRangeSet(persistedRanges);
+                          << to_string(verified);
+    app_.getLedgerMaster().setPinnedLedgersRangeSet(verified);
 }
 
 void

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -184,9 +184,11 @@ public:
     void
     setPinnedRanges(RangeSet<std::uint32_t> const& ranges) override
     {
-        // Save pinned ranges to database for persistence
-        std::string rangesStr = to_string(ranges);
-        state_db_.setPinnedRanges(rangesStr);
+        // Save pinned ranges to database for persistence. to_string()
+        // renders an empty set as "empty", which from_string() rejects, so
+        // store "" instead: loadPinnedRanges() treats that as nothing pinned.
+        state_db_.setPinnedRanges(
+            ranges.empty() ? std::string{} : to_string(ranges));
     }
 
     void

src/xrpld/app/rdb/backend/detail/Node.cpp

@@ -258,42 +258,50 @@ saveValidatedLedger(
         ledger->info().txHash == ledger->txMap().getHash().as_uint256(),
         "ripple::detail::saveValidatedLedger : transaction hash match");
 
+    bool const pinned = app.getLedgerMaster().isPinned(seq);
+
     // Save the ledger header in the hashed object store
     {
         Serializer s(128);
         s.add32(HashPrefix::ledgerMaster);
         addRaw(ledger->info(), s);
         // Use pinnedLEDGER type for pinned ledgers, hotLEDGER for others
-        auto ledgerType =
-            app.getLedgerMaster().isPinned(seq) ? pinnedLEDGER : hotLEDGER;
+        auto ledgerType = pinned ? pinnedLEDGER : hotLEDGER;
         app.getNodeStore().store(
             ledgerType, std::move(s.modData()), ledger->info().hash, seq);
     }
 
+    // Pinned (catalogue) ledgers are saved in bulk: they are never cached
+    // or published from here, so their AcceptedLedger skips the per-tx JSON
+    // and is not built at all when there are no tx tables to fill.
     std::shared_ptr<AcceptedLedger> aLedger;
-    try
+    if (!pinned || app.config().useTxTables())
     {
-        aLedger = app.getAcceptedLedgerCache().fetch(ledger->info().hash);
-        if (!aLedger)
+        try
         {
-            aLedger = std::make_shared<AcceptedLedger>(ledger, app);
-
-            // Only cache if the ledger is NOT in the pinned range
-            if (!app.getLedgerMaster().isPinned(ledger->info().seq))
+            aLedger = app.getAcceptedLedgerCache().fetch(ledger->info().hash);
+            if (!aLedger)
             {
-                app.getAcceptedLedgerCache().canonicalize_replace_client(
-                    ledger->info().hash, aLedger);
+                aLedger = std::make_shared<AcceptedLedger>(
+                    ledger, app, /*withJson=*/!pinned);
+
+                // Only cache if the ledger is NOT in the pinned range
+                if (!pinned)
+                {
+                    app.getAcceptedLedgerCache().canonicalize_replace_client(
+                        ledger->info().hash, aLedger);
+                }
             }
         }
-    }
-    catch (std::exception const&)
-    {
-        JLOG(j.warn()) << "An accepted ledger was missing nodes";
-        app.getLedgerMaster().failedSave(seq, ledger->info().hash);
-        // Clients can now trust the database for information about this
-        // ledger sequence.
-        app.pendingSaves().finishWork(seq);
-        return false;
+        catch (std::exception const&)
+        {
+            JLOG(j.warn()) << "An accepted ledger was missing nodes";
+            app.getLedgerMaster().failedSave(seq, ledger->info().hash);
+            // Clients can now trust the database for information about this
+            // ledger sequence.
+            app.pendingSaves().finishWork(seq);
+            return false;
+        }
     }
 
     {

src/xrpld/net/detail/RPCCall.cpp

@@ -851,6 +851,22 @@ private:
         return jvRequest;
     }
 
+    // catalogue_unpin <ledger_index_min> <ledger_index_max>
+    Json::Value
+    parseCatalogueUnpin(Json::Value const& jvParams)
+    {
+        Json::Value jvRequest(Json::objectValue);
+
+        if (jvParams.size() >= 2)
+        {
+            // asUInt() throws on non-numeric input rather than yielding 0
+            jvRequest[jss::ledger_index_min] = jvParams[0u].asUInt();
+            jvRequest[jss::ledger_index_max] = jvParams[1u].asUInt();
+        }
+
+        return jvRequest;
+    }
+
     // catalogue_status - no parameters required
     Json::Value
     parseCatalogueStatus(Json::Value const& jvParams)
@@ -1470,6 +1486,7 @@ public:
             {"catalogue_create", &RPCParser::parseCatalogueCreate, 3, 4},
             {"catalogue_load", &RPCParser::parseCatalogueLoad, 1, 2},
             {"catalogue_status", &RPCParser::parseCatalogueStatus, 0, 0},
+            {"catalogue_unpin", &RPCParser::parseCatalogueUnpin, 2, 2},
             {"channel_authorize", &RPCParser::parseChannelAuthorize, 3, 4},
             {"channel_verify", &RPCParser::parseChannelVerify, 4, 4},
             {"connect", &RPCParser::parseConnect, 1, 2},

src/xrpld/nodestore/Backend.h

@@ -121,6 +121,22 @@ public:
     virtual void
     sync() = 0;
 
+    /** Make every object previously passed to store() or storeBatch()
+        durable before returning.
+
+        @return true if, on return, those writes are as durable as this
+                backend ever makes them (trivially so for in-memory
+                backends); false if the backend cannot do this on demand and
+                they become durable later on its own schedule (e.g. NuDB's
+                commit thread).
+    */
+    virtual bool
+    syncDurable()
+    {
+        sync();
+        return false;
+    }
+
     /** Visit every object in the database
         This is usually called during import.
         @note This routine will not be called concurrently with itself

src/xrpld/nodestore/backend/MemoryFactory.cpp

@@ -186,6 +186,13 @@ public:
     {
     }
 
+    bool
+    syncDurable() override
+    {
+        // Nothing here outlives the process, so there is nothing to wait for.
+        return true;
+    }
+
     void
     for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
     {

src/xrpld/nodestore/backend/NuDBFactory.cpp

@@ -291,6 +291,9 @@ public:
     void
     sync() override
     {
+        // NuDB commits on its context thread (about once a second) and
+        // offers no public barrier while that thread runs, so this cannot
+        // force durability; syncDurable() therefore reports false.
     }
 
     void

src/xrpld/nodestore/backend/NullFactory.cpp

@@ -81,6 +81,13 @@ public:
     {
     }
 
+    bool
+    syncDurable() override
+    {
+        // Nothing here outlives the process, so there is nothing to wait for.
+        return true;
+    }
+
     void
     for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
     {

src/xrpld/nodestore/backend/RWDBFactory.cpp

@@ -164,6 +164,13 @@ public:
     {
     }
 
+    bool
+    syncDurable() override
+    {
+        // Nothing here outlives the process, so there is nothing to wait for.
+        return true;
+    }
+
     void
     for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
     {

src/xrpld/nodestore/backend/RocksDBFactory.cpp

@@ -383,6 +383,33 @@ public:
     void
     sync() override
     {
+        (void)syncDurable();
+    }
+
+    bool
+    syncDurable() override
+    {
+        // Drain anything queued through store() into RocksDB, then flush
+        // and fsync the WAL so every write so far survives a crash.
+        m_batch.waitForWriting();
+        if (!m_db)
+            return false;
+
+        auto const ret = m_db->FlushWAL(/*sync=*/true);
+        if (ret.ok())
+            return true;
+
+        // Some user-supplied options (e.g. allow_mmap_writes) make WAL sync
+        // unavailable. Report "not durable on demand" rather than failing:
+        // callers then fall back to the non-durable checkpoint path.
+        if (ret.IsNotSupported())
+        {
+            JLOG(m_journal.warn())
+                << "RocksDB cannot sync its WAL on demand: " << ret.ToString();
+            return false;
+        }
+
+        Throw<std::runtime_error>("sync failed: " + ret.ToString());
     }
 
     void

src/xrpld/nodestore/detail/BatchWriter.h

@@ -74,13 +74,16 @@ public:
     int
     getWriteLoad();
 
+    /** Block until every object passed to store() so far has been handed
+        to the callback's writeBatch(). */
+    void
+    waitForWriting();
+
 private:
     void
     performScheduledTask() override;
     void
     writeBatch();
-    void
-    waitForWriting();
 
 private:
     using LockType = std::recursive_mutex;

src/xrpld/nodestore/detail/DatabasePinnedImp.cpp

@@ -91,6 +91,36 @@ DatabasePinnedImp::store(
     }
 }
 
+void
+DatabasePinnedImp::storePinnedBatch(Batch const& batch)
+{
+    if (batch.empty())
+        return;
+
+    std::uint64_t bytes = 0;
+    for (auto const& obj : batch)
+    {
+        XRPL_ASSERT(
+            obj && !isPinnedType(obj->getType()),
+            "ripple::NodeStore::DatabasePinnedImp::storePinnedBatch : "
+            "serializable type");
+        bytes += obj->getData().size();
+    }
+
+    // The persistent backends allowed for pinned_type (NuDB, RocksDB, and
+    // rwdb/memory in standalone) are safe to write from several threads,
+    // so this may overlap a BatchWriter flush of single store() calls.
+    persistent_->storeBatch(batch);
+    pinnedStoreCount_ += batch.size();
+    storeStats(batch.size(), bytes);
+}
+
+bool
+DatabasePinnedImp::syncPinned()
+{
+    return persistent_->syncDurable();
+}
+
 std::shared_ptr<NodeObject>
 DatabasePinnedImp::fetchNodeObject(
     uint256 const& hash,
@@ -112,7 +142,11 @@ DatabasePinnedImp::fetchNodeObject(
         {
             // Try persistent backend first (pinned data)
             if (auto obj = tryPersistent(hash, fetchReport))
+            {
+                if (duplicate)
+                    copyToRotating(*obj, hash, ledgerSeq);
                 return obj;
+            }
 
             // Fallback: try rotating backends
             return rotating_.fetchNodeObject(
@@ -130,7 +164,30 @@ DatabasePinnedImp::fetchNodeObject(
         return obj;
 
     // Fallback: try persistent backend (pinned data)
-    return tryPersistent(hash, fetchReport);
+    auto obj = tryPersistent(hash, fetchReport);
+
+    // `duplicate` is set by online_delete's copy of the current validated
+    // state into the new writable backend. A hot ledger can share nodes
+    // that exist only in persistent storage (e.g. after catalogue_load
+    // switched the LCL to its last ledger); copy those too, so the hot
+    // store becomes self-contained after one rotation and pinned history
+    // can be unpinned and its store discarded without breaking live state.
+    // The pinned-first path above does the same, so the copy happens
+    // whichever backend is tried first.
+    if (obj && duplicate)
+        copyToRotating(*obj, hash, ledgerSeq);
+
+    return obj;
+}
+
+void
+DatabasePinnedImp::copyToRotating(
+    NodeObject const& obj,
+    uint256 const& hash,
+    std::uint32_t ledgerSeq)
+{
+    Blob data(obj.getData());
+    rotating_.store(obj.getType(), std::move(data), hash, ledgerSeq);
 }
 
 void
@@ -262,8 +319,8 @@ DatabasePinnedImp::tryPersistent(uint256 const& hash, FetchReport& fetchReport)
     if (status == ok && nodeObject)
     {
         fetchReport.wasFound = true;
-        // Note: We do NOT copy pinned data to rotating storage even if
-        // duplicate=true. Pinned data stays in persistent storage.
+        // Copying into rotating storage for duplicate=true is handled by
+        // the caller (see fetchNodeObject's default path).
     }
     return nodeObject;
 }

src/xrpld/nodestore/detail/DatabasePinnedImp.h

@@ -121,6 +121,20 @@ public:
         std::uint32_t ledgerSeq) override;
     void
     sync() override;
+
+    /** Write node objects of pinned history straight to the persistent
+        backend in one batch. Objects must carry hot (serializable) types.
+        Used by catalogue_load, which builds whole batches per ledger.
+    */
+    void
+    storePinnedBatch(Batch const& batch);
+
+    /** Make everything written to the persistent backend durable.
+        @return false if the backend cannot guarantee that on demand.
+    */
+    bool
+    syncPinned();
+
     bool
     storeLedger(std::shared_ptr<Ledger const> const& srcLedger) override;
     void
@@ -141,6 +155,14 @@ private:
     bool
     likelyPinned(std::uint32_t ledgerSeq) const;
 
+    /** Copy an object found only in persistent storage into the writable
+        rotating backend (the `duplicate` path of fetchNodeObject). */
+    void
+    copyToRotating(
+        NodeObject const& obj,
+        uint256 const& hash,
+        std::uint32_t ledgerSeq);
+
     std::shared_ptr<NodeObject>
     tryPersistent(uint256 const& hash, FetchReport& fetchReport);
 

src/xrpld/rpc/detail/CatalogueStream.cpp

@@ -213,13 +213,19 @@ deserializeSHAMapFromStream(
     CatalogueInputStream& stream,
     SHAMapNodeType nodeType,
     NodeObjectType flushType,
+    SHAMap::FlushSink const* sink,
     bool allowRemoval,
     beast::Journal const& j)
 {
     try
     {
-        auto deserializeLeaf = [&shaMap, &stream, &j, nodeType, allowRemoval](
-                                   SHAMapNodeType& parsedType) -> bool {
+        // Reused across leaves: resize() keeps capacity, so steady-state
+        // parsing does no per-leaf allocation beyond the SHAMapItem itself.
+        std::vector<std::uint8_t> data;
+
+        auto deserializeLeaf =
+            [&shaMap, &stream, &j, &data, nodeType, allowRemoval](
+                SHAMapNodeType& parsedType) -> bool {
             stream.read(reinterpret_cast<char*>(&parsedType), 1);
 
             if (parsedType == SHAMapNodeType::tnTERMINAL)
@@ -248,7 +254,8 @@ deserializeSHAMapFromStream(
                     return false;
                 }
 
-                if (!shaMap.hasItem(key))
+                // delItem walks once and reports absence itself.
+                if (!shaMap.delItem(key))
                 {
                     JLOG(j.error())
                         << "Deserialization: removal of key " << to_string(key)
@@ -256,7 +263,6 @@ deserializeSHAMapFromStream(
                     return false;
                 }
 
-                shaMap.delItem(key);
                 return true;
             }
 
@@ -279,7 +285,7 @@ deserializeSHAMapFromStream(
                 return false;
             }
 
-            std::vector<std::uint8_t> data(size);
+            data.resize(size);
             stream.read(reinterpret_cast<char*>(data.data()), size);
 
             if (stream.fail())
@@ -290,11 +296,8 @@ deserializeSHAMapFromStream(
                 return false;
             }
 
-            auto item = make_shamapitem(key, makeSlice(data));
-            if (shaMap.hasItem(key))
-                return shaMap.updateGiveItem(nodeType, std::move(item));
-
-            return shaMap.addGiveItem(nodeType, std::move(item));
+            auto item = make_shamapitem(key, Slice(data.data(), size));
+            return shaMap.upsertGiveItem(nodeType, std::move(item));
         };
 
         auto lastParsed = SHAMapNodeType::tnREMOVE;
@@ -308,7 +311,10 @@ deserializeSHAMapFromStream(
             return false;
         }
 
-        shaMap.flushDirty(flushType);
+        if (sink)
+            shaMap.flushDirty(flushType, *sink);
+        else
+            shaMap.flushDirty(flushType);
         return true;
     }
     catch (std::exception const& e)
@@ -345,7 +351,13 @@ deserializeStateMapFromStream(
     beast::Journal const& j)
 {
     return deserializeSHAMapFromStream(
-        stateMap, stream, SHAMapNodeType::tnACCOUNT_STATE, flushType, true, j);
+        stateMap,
+        stream,
+        SHAMapNodeType::tnACCOUNT_STATE,
+        flushType,
+        nullptr,
+        true,
+        j);
 }
 
 bool
@@ -356,7 +368,47 @@ deserializeTxMapFromStream(
     beast::Journal const& j)
 {
     return deserializeSHAMapFromStream(
-        txMap, stream, SHAMapNodeType::tnTRANSACTION_MD, flushType, false, j);
+        txMap,
+        stream,
+        SHAMapNodeType::tnTRANSACTION_MD,
+        flushType,
+        nullptr,
+        false,
+        j);
+}
+
+bool
+deserializeStateMapFromStream(
+    SHAMap& stateMap,
+    CatalogueInputStream& stream,
+    SHAMap::FlushSink const& sink,
+    beast::Journal const& j)
+{
+    return deserializeSHAMapFromStream(
+        stateMap,
+        stream,
+        SHAMapNodeType::tnACCOUNT_STATE,
+        hotACCOUNT_NODE,
+        &sink,
+        true,
+        j);
+}
+
+bool
+deserializeTxMapFromStream(
+    SHAMap& txMap,
+    CatalogueInputStream& stream,
+    SHAMap::FlushSink const& sink,
+    beast::Journal const& j)
+{
+    return deserializeSHAMapFromStream(
+        txMap,
+        stream,
+        SHAMapNodeType::tnTRANSACTION_MD,
+        hotTRANSACTION_NODE,
+        &sink,
+        false,
+        j);
 }
 
 }  // namespace RPC

src/xrpld/rpc/detail/CatalogueStream.h

@@ -59,6 +59,24 @@ deserializeTxMapFromStream(
     NodeObjectType flushType,
     beast::Journal const& j);
 
+/** As above, but dirty nodes are flushed to `sink` (tagged with
+    hotACCOUNT_NODE / hotTRANSACTION_NODE) instead of the nodestore, and are
+    not canonicalized into the TreeNodeCache. See SHAMap::flushDirty(t, sink).
+*/
+bool
+deserializeStateMapFromStream(
+    SHAMap& stateMap,
+    CatalogueInputStream& stream,
+    SHAMap::FlushSink const& sink,
+    beast::Journal const& j);
+
+bool
+deserializeTxMapFromStream(
+    SHAMap& txMap,
+    CatalogueInputStream& stream,
+    SHAMap::FlushSink const& sink,
+    beast::Journal const& j);
+
 }  // namespace RPC
 }  // namespace ripple
 

src/xrpld/rpc/detail/Handler.cpp

@@ -199,6 +199,7 @@ Handler const handlerArray[]{
     {"catalogue_create", byRef(&doCatalogueCreate), Role::ADMIN, NO_CONDITION},
     {"catalogue_status", byRef(&doCatalogueStatus), Role::ADMIN, NO_CONDITION},
     {"catalogue_load", byRef(&doCatalogueLoad), Role::ADMIN, NO_CONDITION},
+    {"catalogue_unpin", byRef(&doCatalogueUnpin), Role::ADMIN, NO_CONDITION},
 };
 
 class HandlerTable

src/xrpld/rpc/handlers/Handlers.h

@@ -184,6 +184,8 @@ Json::Value
 doCatalogueStatus(RPC::JsonContext&);
 Json::Value
 doCatalogueLoad(RPC::JsonContext&);
+Json::Value
+doCatalogueUnpin(RPC::JsonContext&);
 }  // namespace ripple
 
 #endif

src/xrpld/shamap/SHAMap.h

@@ -36,6 +36,7 @@
 #include <xrpl/beast/utility/instrumentation.h>
 #include <boost/iostreams/filtering_stream.hpp>
 #include <cassert>
+#include <functional>
 #include <stack>
 #include <vector>
 
@@ -212,6 +213,19 @@ public:
         SHAMapNodeType type,
         boost::intrusive_ptr<SHAMapItem const> item);
 
+    /** Replace the item with the same key if present, otherwise add it.
+
+        Equivalent to `hasItem(key) ? updateGiveItem(..) : addGiveItem(..)`
+        but walks the tree once when the key already exists.
+
+        @return false if the key exists with a different leaf type, or if
+                the add fails.
+    */
+    bool
+    upsertGiveItem(
+        SHAMapNodeType type,
+        boost::intrusive_ptr<SHAMapItem const> item);
+
     // Save a copy if you need to extend the life
     // of the SHAMapItem beyond this SHAMap
     boost::intrusive_ptr<SHAMapItem const> const&
@@ -367,6 +381,26 @@ public:
     int
     flushDirty(NodeObjectType t);
 
+    /** Receives each node written by flushDirty(t, sink): the type passed
+        to flushDirty, the node serialized with its hash prefix, and its
+        hash.
+    */
+    using FlushSink =
+        std::function<void(NodeObjectType, Blob&&, uint256 const&)>;
+
+    /** Flush modified nodes to `sink` instead of the nodestore and convert
+        them to shared.
+
+        Unlike flushDirty(t), written nodes are NOT canonicalized into the
+        family's TreeNodeCache. Bulk writers of history (catalogue_load)
+        must use this: every cached inner node holds strong references to
+        its children, so canonicalizing pins every superseded node version
+        in memory until the cache sweeps it, which on a fast import grows
+        without practical bound.
+    */
+    int
+    flushDirty(NodeObjectType t, FlushSink const& sink);
+
     void
     walkMap(std::vector<SHAMapMissingNode>& missingNodes, int maxMissing) const;
     bool
@@ -449,9 +483,13 @@ private:
     std::shared_ptr<Node>
     preFlushNode(std::shared_ptr<Node> node) const;
 
-    /** write and canonicalize modified node */
+    /** write and canonicalize modified node; if `sink` is set, hand the
+        serialized node to it instead and do not canonicalize */
     std::shared_ptr<SHAMapTreeNode>
-    writeNode(NodeObjectType t, std::shared_ptr<SHAMapTreeNode> node) const;
+    writeNode(
+        NodeObjectType t,
+        std::shared_ptr<SHAMapTreeNode> node,
+        FlushSink const* sink = nullptr) const;
 
     // returns the first item at or below this node
     SHAMapLeafNode*
@@ -534,7 +572,10 @@ private:
         Delta& differences,
         int& maxCount) const;
     int
-    walkSubTree(bool doWrite, NodeObjectType t);
+    walkSubTree(
+        bool doWrite,
+        NodeObjectType t,
+        FlushSink const* sink = nullptr);
 
     // Structure to track information about call to
     // getMissingNodes while it's in progress

src/xrpld/shamap/detail/SHAMap.cpp

@@ -1059,6 +1059,42 @@ SHAMap::updateGiveItem(
     return true;
 }
 
+bool
+SHAMap::upsertGiveItem(
+    SHAMapNodeType type,
+    boost::intrusive_ptr<SHAMapItem const> item)
+{
+    XRPL_ASSERT(
+        state_ != SHAMapState::Immutable,
+        "ripple::SHAMap::upsertGiveItem : not immutable");
+
+    uint256 const tag = item->key();
+
+    SharedPtrNodeStack stack;
+    auto const leaf = walkTowardsKey(tag, &stack);
+
+    if (!leaf || leaf->peekItem()->key() != tag)
+        return addGiveItem(type, std::move(item));
+
+    // From here this mirrors updateGiveItem, reusing the walk above.
+    auto node = std::static_pointer_cast<SHAMapLeafNode>(stack.top().first);
+    auto const nodeID = stack.top().second;
+    stack.pop();
+
+    if (node->getType() != type)
+    {
+        JLOG(journal_.fatal()) << "SHAMap::upsertGiveItem: cross-type change!";
+        return false;
+    }
+
+    node = unshareNode(std::move(node), nodeID);
+
+    if (node->setItem(std::move(item)))
+        dirtyUp(stack, tag, node);
+
+    return true;
+}
+
 bool
 SHAMap::fetchRoot(SHAMapHash const& hash, SHAMapSyncFilter* filter)
 {
@@ -1108,10 +1144,24 @@ SHAMap::fetchRoot(SHAMapHash const& hash, SHAMapSyncFilter* filter)
           first call SHAMapTreeNode::unshare().
  */
 std::shared_ptr<SHAMapTreeNode>
-SHAMap::writeNode(NodeObjectType t, std::shared_ptr<SHAMapTreeNode> node) const
+SHAMap::writeNode(
+    NodeObjectType t,
+    std::shared_ptr<SHAMapTreeNode> node,
+    FlushSink const* sink) const
 {
     XRPL_ASSERT(
         node->cowid() == 0, "ripple::SHAMap::writeNode : valid input node");
+
+    if (sink)
+    {
+        // Bulk-write path: no TreeNodeCache, no direct store. The caller
+        // owns persistence (see flushDirty(t, sink)).
+        Serializer s;
+        node->serializeWithPrefix(s);
+        (*sink)(t, std::move(s.modData()), node->getHash().as_uint256());
+        return node;
+    }
+
     XRPL_ASSERT(backed_, "ripple::SHAMap::writeNode : is backed");
 
     canonicalize(node->getHash(), node);
@@ -1159,10 +1209,19 @@ SHAMap::flushDirty(NodeObjectType t)
 }
 
 int
-SHAMap::walkSubTree(bool doWrite, NodeObjectType t)
+SHAMap::flushDirty(NodeObjectType t, FlushSink const& sink)
+{
+    // The sink decides where nodes go, so this writes whether or not the
+    // map is backed.
+    return walkSubTree(true, t, &sink);
+}
+
+int
+SHAMap::walkSubTree(bool doWrite, NodeObjectType t, FlushSink const* sink)
 {
     XRPL_ASSERT(
-        !doWrite || backed_, "ripple::SHAMap::walkSubTree : valid input");
+        !doWrite || backed_ || sink,
+        "ripple::SHAMap::walkSubTree : valid input");
 
     int flushed = 0;
 
@@ -1176,7 +1235,7 @@ SHAMap::walkSubTree(bool doWrite, NodeObjectType t)
         root_->unshare();
 
         if (doWrite)
-            root_ = writeNode(t, std::move(root_));
+            root_ = writeNode(t, std::move(root_), sink);
 
         return 1;
     }
@@ -1245,7 +1304,7 @@ SHAMap::walkSubTree(bool doWrite, NodeObjectType t)
                         child->unshare();
 
                         if (doWrite)
-                            child = writeNode(t, std::move(child));
+                            child = writeNode(t, std::move(child), sink);
 
                         node->shareChild(branch, child);
                     }
@@ -1261,7 +1320,7 @@ SHAMap::walkSubTree(bool doWrite, NodeObjectType t)
 
         if (doWrite)
             node = std::static_pointer_cast<SHAMapInnerNode>(
-                writeNode(t, std::move(node)));
+                writeNode(t, std::move(node), sink));
 
         ++flushed;