Restore freshening caches of tree node cache

This changes switches from the GitHub-managed Windows runners to self-hosted runners to significantly reduce build time.

.github/scripts/strategy-matrix/windows.json

@@ -2,7 +2,7 @@
   "architecture": [
     {
       "platform": "windows/amd64",
-      "runner": ["windows-latest"]
+      "runner": ["self-hosted", "Windows", "devbox"]
     }
   ],
   "os": [

.github/workflows/on-pr.yml

@@ -92,12 +92,12 @@ jobs:
 
   check-levelization:
     needs: should-run
-    if: needs.should-run.outputs.go == 'true'
+    if: ${{ needs.should-run.outputs.go == 'true' }}
     uses: ./.github/workflows/check-levelization.yml
 
   build-test:
     needs: should-run
-    if: needs.should-run.outputs.go == 'true'
+    if: ${{ needs.should-run.outputs.go == 'true' }}
     uses: ./.github/workflows/build-test.yml
     strategy:
       matrix:
@@ -111,7 +111,7 @@ jobs:
     needs:
       - should-run
       - build-test
-    if: needs.should-run.outputs.go == 'true'
+    if: ${{ needs.should-run.outputs.go == 'true' && contains(fromJSON('["release", "master"]'), github.ref_name) }}
     uses: ./.github/workflows/notify-clio.yml
     secrets:
       clio_notify_token: ${{ secrets.CLIO_NOTIFY_TOKEN }}

.github/workflows/upload-conan-deps.yml

@@ -34,6 +34,10 @@ on:
       - conanfile.py
       - conan.lock
 
+env:
+  CONAN_REMOTE_NAME: xrplf
+  CONAN_REMOTE_URL: https://conan.ripplex.io
+
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true
@@ -67,6 +71,9 @@ jobs:
 
       - name: Setup Conan
         uses: ./.github/actions/setup-conan
+        with:
+          conan_remote_name: ${{ env.CONAN_REMOTE_NAME }}
+          conan_remote_url: ${{ env.CONAN_REMOTE_URL }}
 
       - name: Build dependencies
         uses: ./.github/actions/build-deps
@@ -75,10 +82,10 @@ jobs:
           build_type: ${{ matrix.build_type }}
           force_build: ${{ github.event_name == 'schedule' || github.event.inputs.force_source_build == 'true' }}
 
-      - name: Login to Conan
-        if: github.repository_owner == 'XRPLF' && github.event_name != 'pull_request'
-        run: conan remote login -p ${{ secrets.CONAN_PASSWORD }} ${{ inputs.conan_remote_name }} ${{ secrets.CONAN_USERNAME }}
+      - name: Log into Conan remote
+        if: ${{ github.repository_owner == 'XRPLF' && github.event_name != 'pull_request' }}
+        run: conan remote login ${{ env.CONAN_REMOTE_NAME }} "${{ secrets.CONAN_REMOTE_USERNAME }}" --password "${{ secrets.CONAN_REMOTE_PASSWORD }}"
 
       - name: Upload Conan packages
-        if: github.repository_owner == 'XRPLF' && github.event_name != 'pull_request' && github.event_name != 'schedule'
-        run: conan upload "*" -r=${{ inputs.conan_remote_name }} --confirm ${{ github.event.inputs.force_upload == 'true' && '--force' || '' }}
+        if: ${{ github.repository_owner == 'XRPLF' && github.event_name != 'pull_request' && github.event_name != 'schedule' }}
+        run: conan upload "*" -r=${{ env.CONAN_REMOTE_NAME }} --confirm ${{ github.event.inputs.force_upload == 'true' && '--force' || '' }}

cfg/rippled-example.cfg

@@ -940,23 +940,7 @@
 #
 #       path                Location to store the database
 #
-#   Optional keys
-#
-#       cache_size          Size of cache for database records. Default is 16384.
-#                           Setting this value to 0 will use the default value.
-#
-#       cache_age           Length of time in minutes to keep database records
-#                           cached. Default is 5 minutes. Setting this value to
-#                           0 will use the default value.
-#
-#                           Note: if neither cache_size nor cache_age is
-#                           specified, the cache for database records will not
-#                           be created. If only one of cache_size or cache_age
-#                           is specified, the cache will be created using the
-#                           default value for the unspecified parameter.
-#
-#                           Note: the cache will not be created if online_delete
-#                           is specified.
+#   Optional keys for NuDB and RocksDB:
 #
 #       fast_load           Boolean. If set, load the last persisted ledger
 #                           from disk upon process start before syncing to
@@ -964,8 +948,6 @@
 #                           if sufficient IOPS capacity is available.
 #                           Default 0.
 #
-#   Optional keys for NuDB or RocksDB:
-#
 #       earliest_seq        The default is 32570 to match the XRP ledger
 #                           network's earliest allowed sequence. Alternate
 #                           networks may set this value. Minimum value of 1.

cmake/CodeCoverage.cmake

@@ -104,6 +104,11 @@
 # 2025-08-28, Bronek Kozicki
 #     - fix "At least one COMMAND must be given" CMake warning from policy CMP0175
 #
+# 2025-09-03, Jingchen Wu
+#     - remove the unused function append_coverage_compiler_flags and append_coverage_compiler_flags_to_target
+#     - add a new function add_code_coverage_to_target
+#     - remove some unused code
+#
 # USAGE:
 #
 # 1. Copy this file into your cmake modules path.
@@ -112,10 +117,8 @@
 #    using a CMake option() to enable it just optionally):
 #      include(CodeCoverage)
 #
-# 3. Append necessary compiler flags for all supported source files:
-#      append_coverage_compiler_flags()
-#    Or for specific target:
-#      append_coverage_compiler_flags_to_target(YOUR_TARGET_NAME)
+# 3. Append necessary compiler flags and linker flags for all supported source files:
+#      add_code_coverage_to_target(<target> <PRIVATE|PUBLIC|INTERFACE>)
 #
 # 3.a (OPTIONAL) Set appropriate optimization flags, e.g. -O0, -O1 or -Og
 #
@@ -204,67 +207,69 @@ endforeach()
 
 set(COVERAGE_COMPILER_FLAGS "-g --coverage"
     CACHE INTERNAL "")
+
+set(COVERAGE_CXX_COMPILER_FLAGS "")
+set(COVERAGE_C_COMPILER_FLAGS "")
+set(COVERAGE_CXX_LINKER_FLAGS "")
+set(COVERAGE_C_LINKER_FLAGS "")
+
 if(CMAKE_CXX_COMPILER_ID MATCHES "(GNU|Clang)")
     include(CheckCXXCompilerFlag)
     include(CheckCCompilerFlag)
+    include(CheckLinkerFlag)
+
+    set(COVERAGE_CXX_COMPILER_FLAGS ${COVERAGE_COMPILER_FLAGS})
+    set(COVERAGE_C_COMPILER_FLAGS ${COVERAGE_COMPILER_FLAGS})
+    set(COVERAGE_CXX_LINKER_FLAGS ${COVERAGE_COMPILER_FLAGS})
+    set(COVERAGE_C_LINKER_FLAGS ${COVERAGE_COMPILER_FLAGS})
 
     check_cxx_compiler_flag(-fprofile-abs-path HAVE_cxx_fprofile_abs_path)
     if(HAVE_cxx_fprofile_abs_path)
-        set(COVERAGE_CXX_COMPILER_FLAGS "${COVERAGE_COMPILER_FLAGS} -fprofile-abs-path")
+        set(COVERAGE_CXX_COMPILER_FLAGS "${COVERAGE_CXX_COMPILER_FLAGS} -fprofile-abs-path")
     endif()
 
     check_c_compiler_flag(-fprofile-abs-path HAVE_c_fprofile_abs_path)
     if(HAVE_c_fprofile_abs_path)
-        set(COVERAGE_C_COMPILER_FLAGS "${COVERAGE_COMPILER_FLAGS} -fprofile-abs-path")
+        set(COVERAGE_C_COMPILER_FLAGS "${COVERAGE_C_COMPILER_FLAGS} -fprofile-abs-path")
+    endif()
+
+    check_linker_flag(CXX -fprofile-abs-path HAVE_cxx_linker_fprofile_abs_path)
+    if(HAVE_cxx_linker_fprofile_abs_path)
+        set(COVERAGE_CXX_LINKER_FLAGS "${COVERAGE_CXX_LINKER_FLAGS} -fprofile-abs-path")
+    endif()
+
+    check_linker_flag(C -fprofile-abs-path HAVE_c_linker_fprofile_abs_path)
+    if(HAVE_c_linker_fprofile_abs_path)
+        set(COVERAGE_C_LINKER_FLAGS "${COVERAGE_C_LINKER_FLAGS} -fprofile-abs-path")
     endif()
 
     check_cxx_compiler_flag(-fprofile-update=atomic HAVE_cxx_fprofile_update)
     if(HAVE_cxx_fprofile_update)
-        set(COVERAGE_CXX_COMPILER_FLAGS "${COVERAGE_COMPILER_FLAGS}  -fprofile-update=atomic")
+        set(COVERAGE_CXX_COMPILER_FLAGS "${COVERAGE_CXX_COMPILER_FLAGS}  -fprofile-update=atomic")
     endif()
 
     check_c_compiler_flag(-fprofile-update=atomic HAVE_c_fprofile_update)
     if(HAVE_c_fprofile_update)
-        set(COVERAGE_C_COMPILER_FLAGS "${COVERAGE_COMPILER_FLAGS} -fprofile-update=atomic")
+        set(COVERAGE_C_COMPILER_FLAGS "${COVERAGE_C_COMPILER_FLAGS} -fprofile-update=atomic")
     endif()
-endif()
 
-set(CMAKE_Fortran_FLAGS_COVERAGE
-    ${COVERAGE_COMPILER_FLAGS}
-    CACHE STRING "Flags used by the Fortran compiler during coverage builds."
-    FORCE )
-set(CMAKE_CXX_FLAGS_COVERAGE
-    ${COVERAGE_COMPILER_FLAGS}
-    CACHE STRING "Flags used by the C++ compiler during coverage builds."
-    FORCE )
-set(CMAKE_C_FLAGS_COVERAGE
-    ${COVERAGE_COMPILER_FLAGS}
-    CACHE STRING "Flags used by the C compiler during coverage builds."
-    FORCE )
-set(CMAKE_EXE_LINKER_FLAGS_COVERAGE
-    ""
-    CACHE STRING "Flags used for linking binaries during coverage builds."
-    FORCE )
-set(CMAKE_SHARED_LINKER_FLAGS_COVERAGE
-    ""
-    CACHE STRING "Flags used by the shared libraries linker during coverage builds."
-    FORCE )
-mark_as_advanced(
-    CMAKE_Fortran_FLAGS_COVERAGE
-    CMAKE_CXX_FLAGS_COVERAGE
-    CMAKE_C_FLAGS_COVERAGE
-    CMAKE_EXE_LINKER_FLAGS_COVERAGE
-    CMAKE_SHARED_LINKER_FLAGS_COVERAGE )
+    check_linker_flag(CXX -fprofile-update=atomic HAVE_cxx_linker_fprofile_update)
+    if(HAVE_cxx_linker_fprofile_update)
+        set(COVERAGE_CXX_LINKER_FLAGS "${COVERAGE_CXX_LINKER_FLAGS} -fprofile-update=atomic")
+    endif()
+
+    check_linker_flag(C -fprofile-update=atomic HAVE_c_linker_fprofile_update)
+    if(HAVE_c_linker_fprofile_update)
+        set(COVERAGE_C_LINKER_FLAGS "${COVERAGE_C_LINKER_FLAGS} -fprofile-update=atomic")
+    endif()
+
+endif()
 
 get_property(GENERATOR_IS_MULTI_CONFIG GLOBAL PROPERTY GENERATOR_IS_MULTI_CONFIG)
 if(NOT (CMAKE_BUILD_TYPE STREQUAL "Debug" OR GENERATOR_IS_MULTI_CONFIG))
     message(WARNING "Code coverage results with an optimised (non-Debug) build may be misleading")
 endif() # NOT (CMAKE_BUILD_TYPE STREQUAL "Debug" OR GENERATOR_IS_MULTI_CONFIG)
 
-if(CMAKE_C_COMPILER_ID STREQUAL "GNU" OR CMAKE_Fortran_COMPILER_ID STREQUAL "GNU")
-    link_libraries(gcov)
-endif()
-
 # Defines a target for running and collection code coverage information
 # Builds dependencies, runs the given executable and outputs reports.
 # NOTE! The executable should always have a ZERO as exit code otherwise
@@ -454,18 +459,19 @@ function(setup_target_for_coverage_gcovr)
     )
 endfunction() # setup_target_for_coverage_gcovr
 
-function(append_coverage_compiler_flags)
-    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${COVERAGE_COMPILER_FLAGS}" PARENT_SCOPE)
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${COVERAGE_COMPILER_FLAGS}" PARENT_SCOPE)
-    set(CMAKE_Fortran_FLAGS "${CMAKE_Fortran_FLAGS} ${COVERAGE_COMPILER_FLAGS}" PARENT_SCOPE)
-    message(STATUS "Appending code coverage compiler flags: ${COVERAGE_COMPILER_FLAGS}")
-endfunction() # append_coverage_compiler_flags
+function(add_code_coverage_to_target name scope)
+    separate_arguments(COVERAGE_CXX_COMPILER_FLAGS NATIVE_COMMAND "${COVERAGE_CXX_COMPILER_FLAGS}")
+    separate_arguments(COVERAGE_C_COMPILER_FLAGS NATIVE_COMMAND "${COVERAGE_C_COMPILER_FLAGS}")
+    separate_arguments(COVERAGE_CXX_LINKER_FLAGS NATIVE_COMMAND "${COVERAGE_CXX_LINKER_FLAGS}")
+    separate_arguments(COVERAGE_C_LINKER_FLAGS NATIVE_COMMAND "${COVERAGE_C_LINKER_FLAGS}")
 
-# Setup coverage for specific library
-function(append_coverage_compiler_flags_to_target name)
-    separate_arguments(_flag_list NATIVE_COMMAND "${COVERAGE_COMPILER_FLAGS}")
-    target_compile_options(${name} PRIVATE ${_flag_list})
-    if(CMAKE_C_COMPILER_ID STREQUAL "GNU" OR CMAKE_Fortran_COMPILER_ID STREQUAL "GNU")
-        target_link_libraries(${name} PRIVATE gcov)
-    endif()
-endfunction()
+    # Add compiler options to the target
+    target_compile_options(${name} ${scope}
+            $<$<COMPILE_LANGUAGE:CXX>:${COVERAGE_CXX_COMPILER_FLAGS}>
+            $<$<COMPILE_LANGUAGE:C>:${COVERAGE_C_COMPILER_FLAGS}>)
+
+    target_link_libraries (${name} ${scope}
+            $<$<LINK_LANGUAGE:CXX>:${COVERAGE_CXX_LINKER_FLAGS} gcov>
+            $<$<LINK_LANGUAGE:C>:${COVERAGE_C_LINKER_FLAGS} gcov>
+    )
+endfunction() # add_code_coverage_to_target

cmake/RippledCov.cmake

@@ -36,3 +36,5 @@ setup_target_for_coverage_gcovr(
   EXCLUDE "src/test" "include/xrpl/beast/test" "include/xrpl/beast/unit_test" "${CMAKE_BINARY_DIR}/pb-xrpl.libpb"
   DEPENDENCIES rippled
 )
+
+add_code_coverage_to_target(opts INTERFACE)

cmake/RippledInterface.cmake

@@ -28,15 +28,11 @@ target_compile_options (opts
     $<$<AND:$<BOOL:${is_gcc}>,$<COMPILE_LANGUAGE:CXX>>:-Wsuggest-override>
     $<$<BOOL:${is_gcc}>:-Wno-maybe-uninitialized>
     $<$<BOOL:${perf}>:-fno-omit-frame-pointer>
-    $<$<AND:$<BOOL:${is_gcc}>,$<BOOL:${coverage}>>:-g --coverage -fprofile-abs-path>
-    $<$<AND:$<BOOL:${is_clang}>,$<BOOL:${coverage}>>:-g --coverage>
     $<$<BOOL:${profile}>:-pg>
     $<$<AND:$<BOOL:${is_gcc}>,$<BOOL:${profile}>>:-p>)
 
 target_link_libraries (opts
   INTERFACE
-    $<$<AND:$<BOOL:${is_gcc}>,$<BOOL:${coverage}>>:-g --coverage -fprofile-abs-path>
-    $<$<AND:$<BOOL:${is_clang}>,$<BOOL:${coverage}>>:-g --coverage>
     $<$<BOOL:${profile}>:-pg>
     $<$<AND:$<BOOL:${is_gcc}>,$<BOOL:${profile}>>:-p>)
 

include/xrpl/protocol/LedgerFormats.h

@@ -188,6 +188,15 @@ enum LedgerSpecificFlags {
     lsfMPTCanTransfer = 0x00000020,
     lsfMPTCanClawback = 0x00000040,
 
+    lsfMPTCanMutateCanLock = 0x00000002,
+    lsfMPTCanMutateRequireAuth = 0x00000004,
+    lsfMPTCanMutateCanEscrow = 0x00000008,
+    lsfMPTCanMutateCanTrade = 0x00000010,
+    lsfMPTCanMutateCanTransfer = 0x00000020,
+    lsfMPTCanMutateCanClawback = 0x00000040,
+    lsfMPTCanMutateMetadata = 0x00010000,
+    lsfMPTCanMutateTransferFee = 0x00020000,
+
     // ltMPTOKEN
     lsfMPTAuthorized = 0x00000002,
 

include/xrpl/protocol/TxFlags.h

@@ -151,6 +151,20 @@ constexpr std::uint32_t const tfMPTCanClawback             = lsfMPTCanClawback;
 constexpr std::uint32_t const tfMPTokenIssuanceCreateMask  =
   ~(tfUniversal | tfMPTCanLock | tfMPTRequireAuth | tfMPTCanEscrow | tfMPTCanTrade | tfMPTCanTransfer | tfMPTCanClawback);
 
+// MPTokenIssuanceCreate MutableFlags:
+// Indicating specific fields or flags may be changed after issuance.
+constexpr std::uint32_t const tfMPTCanMutateCanLock = lsfMPTCanMutateCanLock;
+constexpr std::uint32_t const tfMPTCanMutateRequireAuth = lsfMPTCanMutateRequireAuth;
+constexpr std::uint32_t const tfMPTCanMutateCanEscrow = lsfMPTCanMutateCanEscrow;
+constexpr std::uint32_t const tfMPTCanMutateCanTrade = lsfMPTCanMutateCanTrade;
+constexpr std::uint32_t const tfMPTCanMutateCanTransfer = lsfMPTCanMutateCanTransfer;
+constexpr std::uint32_t const tfMPTCanMutateCanClawback = lsfMPTCanMutateCanClawback;
+constexpr std::uint32_t const tfMPTCanMutateMetadata = lsfMPTCanMutateMetadata;
+constexpr std::uint32_t const tfMPTCanMutateTransferFee = lsfMPTCanMutateTransferFee;
+constexpr std::uint32_t const tfMPTokenIssuanceCreateMutableMask =
+  ~(tfMPTCanMutateCanLock | tfMPTCanMutateRequireAuth | tfMPTCanMutateCanEscrow | tfMPTCanMutateCanTrade
+    | tfMPTCanMutateCanTransfer | tfMPTCanMutateCanClawback | tfMPTCanMutateMetadata | tfMPTCanMutateTransferFee);
+
 // MPTokenAuthorize flags:
 constexpr std::uint32_t const tfMPTUnauthorize             = 0x00000001;
 constexpr std::uint32_t const tfMPTokenAuthorizeMask  = ~(tfUniversal | tfMPTUnauthorize);
@@ -161,6 +175,25 @@ constexpr std::uint32_t const tfMPTUnlock                 = 0x00000002;
 constexpr std::uint32_t const tfMPTokenIssuanceSetMask  = ~(tfUniversal | tfMPTLock | tfMPTUnlock);
 constexpr std::uint32_t const tfMPTokenIssuanceSetPermissionMask = ~(tfUniversal | tfMPTLock | tfMPTUnlock);
 
+// MPTokenIssuanceSet MutableFlags:
+// Set or Clear flags.
+constexpr std::uint32_t const tfMPTSetCanLock             = 0x00000001;
+constexpr std::uint32_t const tfMPTClearCanLock           = 0x00000002;
+constexpr std::uint32_t const tfMPTSetRequireAuth         = 0x00000004;
+constexpr std::uint32_t const tfMPTClearRequireAuth       = 0x00000008;
+constexpr std::uint32_t const tfMPTSetCanEscrow           = 0x00000010;
+constexpr std::uint32_t const tfMPTClearCanEscrow         = 0x00000020;
+constexpr std::uint32_t const tfMPTSetCanTrade            = 0x00000040;
+constexpr std::uint32_t const tfMPTClearCanTrade          = 0x00000080;
+constexpr std::uint32_t const tfMPTSetCanTransfer         = 0x00000100;
+constexpr std::uint32_t const tfMPTClearCanTransfer       = 0x00000200;
+constexpr std::uint32_t const tfMPTSetCanClawback         = 0x00000400;
+constexpr std::uint32_t const tfMPTClearCanClawback       = 0x00000800;
+constexpr std::uint32_t const tfMPTokenIssuanceSetMutableMask = ~(tfMPTSetCanLock | tfMPTClearCanLock |
+    tfMPTSetRequireAuth | tfMPTClearRequireAuth | tfMPTSetCanEscrow | tfMPTClearCanEscrow |
+    tfMPTSetCanTrade | tfMPTClearCanTrade | tfMPTSetCanTransfer | tfMPTClearCanTransfer |
+    tfMPTSetCanClawback | tfMPTClearCanClawback);
+
 // MPTokenIssuanceDestroy flags:
 constexpr std::uint32_t const tfMPTokenIssuanceDestroyMask  = ~tfUniversal;
 

include/xrpl/protocol/detail/features.macro

@@ -32,6 +32,8 @@
 // If you add an amendment here, then do not forget to increment `numFeatures`
 // in include/xrpl/protocol/Feature.h.
 
+XRPL_FEATURE(DynamicMPT,                 Supported::no,  VoteBehavior::DefaultNo)
+XRPL_FIX    (TokenEscrowV1,              Supported::yes, VoteBehavior::DefaultNo)
 XRPL_FIX    (DelegateV1_1,               Supported::no,  VoteBehavior::DefaultNo)
 XRPL_FIX    (PriceOracleOrder,           Supported::no,  VoteBehavior::DefaultNo)
 XRPL_FIX    (MPTDeliveredAmount,         Supported::no,  VoteBehavior::DefaultNo)

include/xrpl/protocol/detail/ledger_entries.macro

@@ -412,6 +412,7 @@ LEDGER_ENTRY(ltMPTOKEN_ISSUANCE, 0x007e, MPTokenIssuance, mpt_issuance, ({
     {sfPreviousTxnID,            soeREQUIRED},
     {sfPreviousTxnLgrSeq,        soeREQUIRED},
     {sfDomainID,                 soeOPTIONAL},
+    {sfMutableFlags,             soeDEFAULT},
 }))
 
 /** A ledger object which tracks MPToken

include/xrpl/protocol/detail/sfields.macro

@@ -114,6 +114,7 @@ TYPED_SFIELD(sfVoteWeight,               UINT32,    48)
 TYPED_SFIELD(sfFirstNFTokenSequence,     UINT32,    50)
 TYPED_SFIELD(sfOracleDocumentID,         UINT32,    51)
 TYPED_SFIELD(sfPermissionValue,          UINT32,    52)
+TYPED_SFIELD(sfMutableFlags,             UINT32,    53)
 
 // 64-bit integers (common)
 TYPED_SFIELD(sfIndexNext,                UINT64,     1)

include/xrpl/protocol/detail/transactions.macro

@@ -548,6 +548,7 @@ TRANSACTION(ttMPTOKEN_ISSUANCE_CREATE, 54, MPTokenIssuanceCreate,
     {sfMaximumAmount, soeOPTIONAL},
     {sfMPTokenMetadata, soeOPTIONAL},
     {sfDomainID, soeOPTIONAL},
+    {sfMutableFlags, soeOPTIONAL},
 }))
 
 /** This transaction type destroys a MPTokensIssuance instance */
@@ -566,6 +567,9 @@ TRANSACTION(ttMPTOKEN_ISSUANCE_SET, 56, MPTokenIssuanceSet,
     {sfMPTokenIssuanceID, soeREQUIRED},
     {sfHolder, soeOPTIONAL},
     {sfDomainID, soeOPTIONAL},
+    {sfMPTokenMetadata, soeOPTIONAL},
+    {sfTransferFee, soeOPTIONAL},
+    {sfMutableFlags, soeOPTIONAL},
 }))
 
 /** This transaction type authorizes a MPToken instance */

src/test/app/EscrowToken_test.cpp

@@ -3501,6 +3501,10 @@ struct EscrowToken_test : public beast::unit_test::suite
             BEAST_EXPECT(
                 transferRate.value == std::uint32_t(1'000'000'000 * 1.25));
 
+            BEAST_EXPECT(mptEscrowed(env, alice, MPT) == 125);
+            BEAST_EXPECT(issuerMPTEscrowed(env, MPT) == 125);
+            BEAST_EXPECT(env.balance(gw, MPT) == MPT(20'000));
+
             // bob can finish escrow
             env(escrow::finish(bob, alice, seq1),
                 escrow::condition(escrow::cb1),
@@ -3510,6 +3514,15 @@ struct EscrowToken_test : public beast::unit_test::suite
 
             BEAST_EXPECT(env.balance(alice, MPT) == preAlice - delta);
             BEAST_EXPECT(env.balance(bob, MPT) == MPT(10'100));
+
+            auto const escrowedWithFix =
+                env.current()->rules().enabled(fixTokenEscrowV1) ? 0 : 25;
+            auto const outstandingWithFix =
+                env.current()->rules().enabled(fixTokenEscrowV1) ? MPT(19'975)
+                                                                 : MPT(20'000);
+            BEAST_EXPECT(mptEscrowed(env, alice, MPT) == escrowedWithFix);
+            BEAST_EXPECT(issuerMPTEscrowed(env, MPT) == escrowedWithFix);
+            BEAST_EXPECT(env.balance(gw, MPT) == outstandingWithFix);
         }
 
         // test locked rate: cancel
@@ -3554,6 +3567,60 @@ struct EscrowToken_test : public beast::unit_test::suite
 
             BEAST_EXPECT(env.balance(alice, MPT) == preAlice);
             BEAST_EXPECT(env.balance(bob, MPT) == preBob);
+            BEAST_EXPECT(env.balance(gw, MPT) == MPT(20'000));
+            BEAST_EXPECT(mptEscrowed(env, alice, MPT) == 0);
+            BEAST_EXPECT(issuerMPTEscrowed(env, MPT) == 0);
+        }
+
+        // test locked rate: issuer is destination
+        {
+            Env env{*this, features};
+            auto const baseFee = env.current()->fees().base;
+            auto const alice = Account("alice");
+            auto const bob = Account("bob");
+            auto const gw = Account("gw");
+
+            MPTTester mptGw(env, gw, {.holders = {alice, bob}});
+            mptGw.create(
+                {.transferFee = 25000,
+                 .ownerCount = 1,
+                 .holderCount = 0,
+                 .flags = tfMPTCanEscrow | tfMPTCanTransfer});
+            mptGw.authorize({.account = alice});
+            mptGw.authorize({.account = bob});
+            auto const MPT = mptGw["MPT"];
+            env(pay(gw, alice, MPT(10'000)));
+            env(pay(gw, bob, MPT(10'000)));
+            env.close();
+
+            // alice can create escrow w/ xfer rate
+            auto const preAlice = env.balance(alice, MPT);
+            auto const seq1 = env.seq(alice);
+            auto const delta = MPT(125);
+            env(escrow::create(alice, gw, MPT(125)),
+                escrow::condition(escrow::cb1),
+                escrow::finish_time(env.now() + 1s),
+                fee(baseFee * 150));
+            env.close();
+            auto const transferRate = escrow::rate(env, alice, seq1);
+            BEAST_EXPECT(
+                transferRate.value == std::uint32_t(1'000'000'000 * 1.25));
+
+            BEAST_EXPECT(mptEscrowed(env, alice, MPT) == 125);
+            BEAST_EXPECT(issuerMPTEscrowed(env, MPT) == 125);
+            BEAST_EXPECT(env.balance(gw, MPT) == MPT(20'000));
+
+            // bob can finish escrow
+            env(escrow::finish(gw, alice, seq1),
+                escrow::condition(escrow::cb1),
+                escrow::fulfillment(escrow::fb1),
+                fee(baseFee * 150));
+            env.close();
+
+            BEAST_EXPECT(env.balance(alice, MPT) == preAlice - delta);
+            BEAST_EXPECT(mptEscrowed(env, alice, MPT) == 0);
+            BEAST_EXPECT(issuerMPTEscrowed(env, MPT) == 0);
+            BEAST_EXPECT(env.balance(gw, MPT) == MPT(19'875));
         }
     }
 
@@ -3878,6 +3945,7 @@ public:
         FeatureBitset const all{testable_amendments()};
         testIOUWithFeats(all);
         testMPTWithFeats(all);
+        testMPTWithFeats(all - fixTokenEscrowV1);
     }
 };
 

src/test/app/MPToken_test.cpp

@@ -589,7 +589,8 @@ class MPToken_test : public beast::unit_test::suite
                  .flags = 0x00000008,
                  .err = temINVALID_FLAG});
 
-            if (!features[featureSingleAssetVault])
+            if (!features[featureSingleAssetVault] &&
+                !features[featureDynamicMPT])
             {
                 // test invalid flags - nothing is being changed
                 mptAlice.set(
@@ -623,7 +624,8 @@ class MPToken_test : public beast::unit_test::suite
                      .flags = 0x00000000,
                      .err = temMALFORMED});
 
-                if (!features[featurePermissionedDomains])
+                if (!features[featurePermissionedDomains] ||
+                    !features[featureSingleAssetVault])
                 {
                     // cannot set DomainID since PD is not enabled
                     mptAlice.set(
@@ -631,7 +633,7 @@ class MPToken_test : public beast::unit_test::suite
                          .domainID = uint256(42),
                          .err = temDISABLED});
                 }
-                else
+                else if (features[featureSingleAssetVault])
                 {
                     // cannot set DomainID since Holder is set
                     mptAlice.set(
@@ -2738,6 +2740,882 @@ class MPToken_test : public beast::unit_test::suite
         }
     }
 
+    void
+    testInvalidCreateDynamic(FeatureBitset features)
+    {
+        testcase("invalid MPTokenIssuanceCreate for DynamicMPT");
+
+        using namespace test::jtx;
+        Account const alice("alice");
+
+        // Can not provide MutableFlags when DynamicMPT amendment is not enabled
+        {
+            Env env{*this, features - featureDynamicMPT};
+            MPTTester mptAlice(env, alice);
+            mptAlice.create(
+                {.ownerCount = 0, .mutableFlags = 2, .err = temDISABLED});
+            mptAlice.create(
+                {.ownerCount = 0, .mutableFlags = 0, .err = temDISABLED});
+        }
+
+        // MutableFlags contains invalid values
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice);
+
+            // Value 1 is reserved for MPT lock.
+            mptAlice.create(
+                {.ownerCount = 0, .mutableFlags = 1, .err = temINVALID_FLAG});
+            mptAlice.create(
+                {.ownerCount = 0, .mutableFlags = 17, .err = temINVALID_FLAG});
+            mptAlice.create(
+                {.ownerCount = 0,
+                 .mutableFlags = 65535,
+                 .err = temINVALID_FLAG});
+
+            // MutableFlags can not be 0
+            mptAlice.create(
+                {.ownerCount = 0, .mutableFlags = 0, .err = temINVALID_FLAG});
+        }
+    }
+
+    void
+    testInvalidSetDynamic(FeatureBitset features)
+    {
+        testcase("invalid MPTokenIssuanceSet for DynamicMPT");
+
+        using namespace test::jtx;
+        Account const alice("alice");
+        Account const bob("bob");
+
+        // Can not provide MutableFlags, MPTokenMetadata or TransferFee when
+        // DynamicMPT amendment is not enabled
+        {
+            Env env{*this, features - featureDynamicMPT};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            auto const mptID = makeMptID(env.seq(alice), alice);
+
+            // MutableFlags is not allowed when DynamicMPT is not enabled
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .mutableFlags = 2,
+                 .err = temDISABLED});
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .mutableFlags = 0,
+                 .err = temDISABLED});
+
+            // MPTokenMetadata is not allowed when DynamicMPT is not enabled
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .metadata = "test",
+                 .err = temDISABLED});
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .metadata = "",
+                 .err = temDISABLED});
+
+            // TransferFee is not allowed when DynamicMPT is not enabled
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .transferFee = 100,
+                 .err = temDISABLED});
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .transferFee = 0,
+                 .err = temDISABLED});
+        }
+
+        // Can not provide holder when MutableFlags, MPTokenMetadata or
+        // TransferFee is present
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            auto const mptID = makeMptID(env.seq(alice), alice);
+
+            // Holder is not allowed when MutableFlags is present
+            mptAlice.set(
+                {.account = alice,
+                 .holder = bob,
+                 .id = mptID,
+                 .mutableFlags = 2,
+                 .err = temMALFORMED});
+
+            // Holder is not allowed when MPTokenMetadata is present
+            mptAlice.set(
+                {.account = alice,
+                 .holder = bob,
+                 .id = mptID,
+                 .metadata = "test",
+                 .err = temMALFORMED});
+
+            // Holder is not allowed when TransferFee is present
+            mptAlice.set(
+                {.account = alice,
+                 .holder = bob,
+                 .id = mptID,
+                 .transferFee = 100,
+                 .err = temMALFORMED});
+        }
+
+        // Can not set Flags when MutableFlags, MPTokenMetadata or
+        // TransferFee is present
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .mutableFlags = tfMPTCanMutateMetadata |
+                     tfMPTCanMutateCanLock | tfMPTCanMutateTransferFee});
+
+            // Setting flags is not allowed when MutableFlags is present
+            mptAlice.set(
+                {.account = alice,
+                 .flags = tfMPTCanLock,
+                 .mutableFlags = 2,
+                 .err = temMALFORMED});
+
+            // Setting flags is not allowed when MPTokenMetadata is present
+            mptAlice.set(
+                {.account = alice,
+                 .flags = tfMPTCanLock,
+                 .metadata = "test",
+                 .err = temMALFORMED});
+
+            // setting flags is not allowed when TransferFee is present
+            mptAlice.set(
+                {.account = alice,
+                 .flags = tfMPTCanLock,
+                 .transferFee = 100,
+                 .err = temMALFORMED});
+        }
+
+        // Flags being 0 or tfFullyCanonicalSig is fine
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create(
+                {.transferFee = 10,
+                 .ownerCount = 1,
+                 .flags = tfMPTCanTransfer,
+                 .mutableFlags =
+                     tfMPTCanMutateTransferFee | tfMPTCanMutateMetadata});
+
+            mptAlice.set(
+                {.account = alice,
+                 .flags = 0,
+                 .transferFee = 100,
+                 .metadata = "test"});
+            mptAlice.set(
+                {.account = alice,
+                 .flags = tfFullyCanonicalSig,
+                 .transferFee = 200,
+                 .metadata = "test2"});
+        }
+
+        // Invalid MutableFlags
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            auto const mptID = makeMptID(env.seq(alice), alice);
+
+            for (auto const flags : {10000, 0, 5000})
+            {
+                mptAlice.set(
+                    {.account = alice,
+                     .id = mptID,
+                     .mutableFlags = flags,
+                     .err = temINVALID_FLAG});
+            }
+        }
+
+        // Can not set and clear the same mutable flag
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            auto const mptID = makeMptID(env.seq(alice), alice);
+
+            auto const flagCombinations = {
+                tfMPTSetCanLock | tfMPTClearCanLock,
+                tfMPTSetRequireAuth | tfMPTClearRequireAuth,
+                tfMPTSetCanEscrow | tfMPTClearCanEscrow,
+                tfMPTSetCanTrade | tfMPTClearCanTrade,
+                tfMPTSetCanTransfer | tfMPTClearCanTransfer,
+                tfMPTSetCanClawback | tfMPTClearCanClawback,
+                tfMPTSetCanLock | tfMPTClearCanLock | tfMPTClearCanTrade,
+                tfMPTSetCanTransfer | tfMPTClearCanTransfer |
+                    tfMPTSetCanEscrow | tfMPTClearCanClawback};
+
+            for (auto const& mutableFlags : flagCombinations)
+            {
+                mptAlice.set(
+                    {.account = alice,
+                     .id = mptID,
+                     .mutableFlags = mutableFlags,
+                     .err = temINVALID_FLAG});
+            }
+        }
+
+        // Can not mutate flag which is not mutable
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create({.ownerCount = 1});
+
+            auto const mutableFlags = {
+                tfMPTSetCanLock,
+                tfMPTClearCanLock,
+                tfMPTSetRequireAuth,
+                tfMPTClearRequireAuth,
+                tfMPTSetCanEscrow,
+                tfMPTClearCanEscrow,
+                tfMPTSetCanTrade,
+                tfMPTClearCanTrade,
+                tfMPTSetCanTransfer,
+                tfMPTClearCanTransfer,
+                tfMPTSetCanClawback,
+                tfMPTClearCanClawback};
+
+            for (auto const& mutableFlag : mutableFlags)
+            {
+                mptAlice.set(
+                    {.account = alice,
+                     .mutableFlags = mutableFlag,
+                     .err = tecNO_PERMISSION});
+            }
+        }
+
+        // Metadata exceeding max length
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create(
+                {.ownerCount = 1, .mutableFlags = tfMPTCanMutateMetadata});
+
+            std::string metadata(maxMPTokenMetadataLength + 1, 'a');
+            mptAlice.set(
+                {.account = alice, .metadata = metadata, .err = temMALFORMED});
+        }
+
+        // Can not mutate metadata when it is not mutable
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create({.ownerCount = 1});
+            mptAlice.set(
+                {.account = alice,
+                 .metadata = "test",
+                 .err = tecNO_PERMISSION});
+        }
+
+        // Transfer fee exceeding the max value
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            auto const mptID = makeMptID(env.seq(alice), alice);
+
+            mptAlice.create(
+                {.ownerCount = 1, .mutableFlags = tfMPTCanMutateTransferFee});
+
+            mptAlice.set(
+                {.account = alice,
+                 .id = mptID,
+                 .transferFee = maxTransferFee + 1,
+                 .err = temBAD_TRANSFER_FEE});
+        }
+
+        // Test setting non-zero transfer fee and clearing MPTCanTransfer at the
+        // same time
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create(
+                {.transferFee = 100,
+                 .ownerCount = 1,
+                 .flags = tfMPTCanTransfer,
+                 .mutableFlags =
+                     tfMPTCanMutateTransferFee | tfMPTCanMutateCanTransfer});
+
+            // Can not set non-zero transfer fee and clear MPTCanTransfer at the
+            // same time
+            mptAlice.set(
+                {.account = alice,
+                 .mutableFlags = tfMPTClearCanTransfer,
+                 .transferFee = 1,
+                 .err = temMALFORMED});
+
+            // Can set transfer fee to zero and clear MPTCanTransfer at the same
+            // time. tfMPTCanTransfer will be cleared and TransferFee field will
+            // be removed.
+            mptAlice.set(
+                {.account = alice,
+                 .mutableFlags = tfMPTClearCanTransfer,
+                 .transferFee = 0});
+            BEAST_EXPECT(!mptAlice.isTransferFeePresent());
+        }
+
+        // Can not set non-zero transfer fee when MPTCanTransfer is not set
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .mutableFlags =
+                     tfMPTCanMutateTransferFee | tfMPTCanMutateCanTransfer});
+
+            mptAlice.set(
+                {.account = alice,
+                 .transferFee = 100,
+                 .err = tecNO_PERMISSION});
+
+            // Can not set transfer fee even when trying to set MPTCanTransfer
+            // at the same time. MPTCanTransfer must be set first, then transfer
+            // fee can be set in a separate transaction.
+            mptAlice.set(
+                {.account = alice,
+                 .mutableFlags = tfMPTSetCanTransfer,
+                 .transferFee = 100,
+                 .err = tecNO_PERMISSION});
+        }
+
+        // Can not mutate transfer fee when it is not mutable
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create(
+                {.transferFee = 10,
+                 .ownerCount = 1,
+                 .flags = tfMPTCanTransfer});
+
+            mptAlice.set(
+                {.account = alice,
+                 .transferFee = 100,
+                 .err = tecNO_PERMISSION});
+
+            mptAlice.set(
+                {.account = alice, .transferFee = 0, .err = tecNO_PERMISSION});
+        }
+
+        // Set some flags mutable. Can not mutate the others
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .mutableFlags = tfMPTCanMutateCanTrade |
+                     tfMPTCanMutateCanTransfer | tfMPTCanMutateMetadata});
+
+            // Can not mutate transfer fee
+            mptAlice.set(
+                {.account = alice,
+                 .transferFee = 100,
+                 .err = tecNO_PERMISSION});
+
+            auto const invalidFlags = {
+                tfMPTSetCanLock,
+                tfMPTClearCanLock,
+                tfMPTSetRequireAuth,
+                tfMPTClearRequireAuth,
+                tfMPTSetCanEscrow,
+                tfMPTClearCanEscrow,
+                tfMPTSetCanClawback,
+                tfMPTClearCanClawback};
+
+            // Can not mutate flags which are not mutable
+            for (auto const& mutableFlag : invalidFlags)
+            {
+                mptAlice.set(
+                    {.account = alice,
+                     .mutableFlags = mutableFlag,
+                     .err = tecNO_PERMISSION});
+            }
+
+            // Can mutate MPTCanTrade
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanTrade});
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTClearCanTrade});
+
+            // Can mutate MPTCanTransfer
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTSetCanTransfer});
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTClearCanTransfer});
+
+            // Can mutate metadata
+            mptAlice.set({.account = alice, .metadata = "test"});
+            mptAlice.set({.account = alice, .metadata = ""});
+        }
+    }
+
+    void
+    testMutateMPT(FeatureBitset features)
+    {
+        testcase("Mutate MPT");
+        using namespace test::jtx;
+
+        Account const alice("alice");
+
+        // Mutate metadata
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice);
+            mptAlice.create(
+                {.metadata = "test",
+                 .ownerCount = 1,
+                 .mutableFlags = tfMPTCanMutateMetadata});
+
+            std::vector<std::string> metadatas = {
+                "mutate metadata",
+                "mutate metadata 2",
+                "mutate metadata 3",
+                "mutate metadata 3",
+                "test",
+                "mutate metadata"};
+
+            for (auto const& metadata : metadatas)
+            {
+                mptAlice.set({.account = alice, .metadata = metadata});
+                BEAST_EXPECT(mptAlice.checkMetadata(metadata));
+            }
+
+            // Metadata being empty will remove the field
+            mptAlice.set({.account = alice, .metadata = ""});
+            BEAST_EXPECT(!mptAlice.isMetadataPresent());
+        }
+
+        // Mutate transfer fee
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice);
+            mptAlice.create(
+                {.transferFee = 100,
+                 .metadata = "test",
+                 .ownerCount = 1,
+                 .flags = tfMPTCanTransfer,
+                 .mutableFlags = tfMPTCanMutateTransferFee});
+
+            for (std::uint16_t const fee : std::initializer_list<std::uint16_t>{
+                     1, 10, 100, 200, 500, 1000, maxTransferFee})
+            {
+                mptAlice.set({.account = alice, .transferFee = fee});
+                BEAST_EXPECT(mptAlice.checkTransferFee(fee));
+            }
+
+            // Setting TransferFee to zero will remove the field
+            mptAlice.set({.account = alice, .transferFee = 0});
+            BEAST_EXPECT(!mptAlice.isTransferFeePresent());
+
+            // Set transfer fee again
+            mptAlice.set({.account = alice, .transferFee = 10});
+            BEAST_EXPECT(mptAlice.checkTransferFee(10));
+        }
+
+        // Test flag toggling
+        {
+            auto testFlagToggle = [&](std::uint32_t createFlags,
+                                      std::uint32_t setFlags,
+                                      std::uint32_t clearFlags) {
+                Env env{*this, features};
+                MPTTester mptAlice(env, alice);
+
+                // Create the MPT object with the specified initial flags
+                mptAlice.create(
+                    {.metadata = "test",
+                     .ownerCount = 1,
+                     .mutableFlags = createFlags});
+
+                // Set and clear the flag multiple times
+                mptAlice.set({.account = alice, .mutableFlags = setFlags});
+                mptAlice.set({.account = alice, .mutableFlags = clearFlags});
+                mptAlice.set({.account = alice, .mutableFlags = clearFlags});
+                mptAlice.set({.account = alice, .mutableFlags = setFlags});
+                mptAlice.set({.account = alice, .mutableFlags = setFlags});
+                mptAlice.set({.account = alice, .mutableFlags = clearFlags});
+                mptAlice.set({.account = alice, .mutableFlags = setFlags});
+                mptAlice.set({.account = alice, .mutableFlags = clearFlags});
+            };
+
+            testFlagToggle(
+                tfMPTCanMutateCanLock, tfMPTCanLock, tfMPTClearCanLock);
+            testFlagToggle(
+                tfMPTCanMutateRequireAuth,
+                tfMPTSetRequireAuth,
+                tfMPTClearRequireAuth);
+            testFlagToggle(
+                tfMPTCanMutateCanEscrow,
+                tfMPTSetCanEscrow,
+                tfMPTClearCanEscrow);
+            testFlagToggle(
+                tfMPTCanMutateCanTrade, tfMPTSetCanTrade, tfMPTClearCanTrade);
+            testFlagToggle(
+                tfMPTCanMutateCanTransfer,
+                tfMPTSetCanTransfer,
+                tfMPTClearCanTransfer);
+            testFlagToggle(
+                tfMPTCanMutateCanClawback,
+                tfMPTSetCanClawback,
+                tfMPTClearCanClawback);
+        }
+    }
+
+    void
+    testMutateCanLock(FeatureBitset features)
+    {
+        testcase("Mutate MPTCanLock");
+        using namespace test::jtx;
+
+        Account const alice("alice");
+        Account const bob("bob");
+
+        // Individual lock
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .holderCount = 0,
+                 .flags = tfMPTCanLock | tfMPTCanTransfer,
+                 .mutableFlags = tfMPTCanMutateCanLock |
+                     tfMPTCanMutateCanTrade | tfMPTCanMutateTransferFee});
+            mptAlice.authorize({.account = bob, .holderCount = 1});
+
+            // Lock bob's mptoken
+            mptAlice.set({.account = alice, .holder = bob, .flags = tfMPTLock});
+
+            // Can mutate the mutable flags and fields
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanLock});
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanLock});
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanLock});
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanTrade});
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTClearCanTrade});
+            mptAlice.set({.account = alice, .transferFee = 200});
+        }
+
+        // Global lock
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .holderCount = 0,
+                 .flags = tfMPTCanLock,
+                 .mutableFlags = tfMPTCanMutateCanLock |
+                     tfMPTCanMutateCanClawback | tfMPTCanMutateMetadata});
+            mptAlice.authorize({.account = bob, .holderCount = 1});
+
+            // Lock issuance
+            mptAlice.set({.account = alice, .flags = tfMPTLock});
+
+            // Can mutate the mutable flags and fields
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanLock});
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanLock});
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanLock});
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTSetCanClawback});
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTClearCanClawback});
+            mptAlice.set({.account = alice, .metadata = "mutate"});
+        }
+
+        // Test lock and unlock after mutating MPTCanLock
+        {
+            Env env{*this, features};
+            MPTTester mptAlice(env, alice, {.holders = {bob}});
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .holderCount = 0,
+                 .flags = tfMPTCanLock,
+                 .mutableFlags = tfMPTCanMutateCanLock |
+                     tfMPTCanMutateCanClawback | tfMPTCanMutateMetadata});
+            mptAlice.authorize({.account = bob, .holderCount = 1});
+
+            // Can lock and unlock
+            mptAlice.set({.account = alice, .flags = tfMPTLock});
+            mptAlice.set({.account = alice, .holder = bob, .flags = tfMPTLock});
+            mptAlice.set({.account = alice, .flags = tfMPTUnlock});
+            mptAlice.set(
+                {.account = alice, .holder = bob, .flags = tfMPTUnlock});
+
+            // Clear lsfMPTCanLock
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanLock});
+
+            // Can not lock or unlock
+            mptAlice.set(
+                {.account = alice,
+                 .flags = tfMPTLock,
+                 .err = tecNO_PERMISSION});
+            mptAlice.set(
+                {.account = alice,
+                 .flags = tfMPTUnlock,
+                 .err = tecNO_PERMISSION});
+            mptAlice.set(
+                {.account = alice,
+                 .holder = bob,
+                 .flags = tfMPTLock,
+                 .err = tecNO_PERMISSION});
+            mptAlice.set(
+                {.account = alice,
+                 .holder = bob,
+                 .flags = tfMPTUnlock,
+                 .err = tecNO_PERMISSION});
+
+            // Set MPTCanLock again
+            mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanLock});
+
+            // Can lock and unlock again
+            mptAlice.set({.account = alice, .flags = tfMPTLock});
+            mptAlice.set({.account = alice, .holder = bob, .flags = tfMPTLock});
+            mptAlice.set({.account = alice, .flags = tfMPTUnlock});
+            mptAlice.set(
+                {.account = alice, .holder = bob, .flags = tfMPTUnlock});
+        }
+    }
+
+    void
+    testMutateRequireAuth(FeatureBitset features)
+    {
+        testcase("Mutate MPTRequireAuth");
+        using namespace test::jtx;
+
+        Env env{*this, features};
+        Account const alice("alice");
+        Account const bob("bob");
+
+        MPTTester mptAlice(env, alice, {.holders = {bob}});
+        mptAlice.create(
+            {.ownerCount = 1,
+             .flags = tfMPTRequireAuth,
+             .mutableFlags = tfMPTCanMutateRequireAuth});
+
+        mptAlice.authorize({.account = bob});
+        mptAlice.authorize({.account = alice, .holder = bob});
+
+        // Pay to bob
+        mptAlice.pay(alice, bob, 1000);
+
+        // Unauthorize bob
+        mptAlice.authorize(
+            {.account = alice, .holder = bob, .flags = tfMPTUnauthorize});
+
+        // Can not pay to bob
+        mptAlice.pay(bob, alice, 100, tecNO_AUTH);
+
+        // Clear RequireAuth
+        mptAlice.set({.account = alice, .mutableFlags = tfMPTClearRequireAuth});
+
+        // Can pay to bob
+        mptAlice.pay(alice, bob, 1000);
+
+        // Set RequireAuth again
+        mptAlice.set({.account = alice, .mutableFlags = tfMPTSetRequireAuth});
+
+        // Can not pay to bob since he is not authorized
+        mptAlice.pay(bob, alice, 100, tecNO_AUTH);
+
+        // Authorize bob again
+        mptAlice.authorize({.account = alice, .holder = bob});
+
+        // Can pay to bob again
+        mptAlice.pay(alice, bob, 100);
+    }
+
+    void
+    testMutateCanEscrow(FeatureBitset features)
+    {
+        testcase("Mutate MPTCanEscrow");
+        using namespace test::jtx;
+        using namespace std::literals;
+
+        Env env{*this, features};
+        auto const baseFee = env.current()->fees().base;
+        auto const alice = Account("alice");
+        auto const bob = Account("bob");
+        auto const carol = Account("carol");
+
+        MPTTester mptAlice(env, alice, {.holders = {carol, bob}});
+        mptAlice.create(
+            {.ownerCount = 1,
+             .holderCount = 0,
+             .flags = tfMPTCanTransfer,
+             .mutableFlags = tfMPTCanMutateCanEscrow});
+        mptAlice.authorize({.account = carol});
+        mptAlice.authorize({.account = bob});
+
+        auto const MPT = mptAlice["MPT"];
+        env(pay(alice, carol, MPT(10'000)));
+        env(pay(alice, bob, MPT(10'000)));
+        env.close();
+
+        // MPTCanEscrow is not enabled
+        env(escrow::create(carol, bob, MPT(3)),
+            escrow::condition(escrow::cb1),
+            escrow::finish_time(env.now() + 1s),
+            fee(baseFee * 150),
+            ter(tecNO_PERMISSION));
+
+        // MPTCanEscrow is enabled now
+        mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanEscrow});
+        env(escrow::create(carol, bob, MPT(3)),
+            escrow::condition(escrow::cb1),
+            escrow::finish_time(env.now() + 1s),
+            fee(baseFee * 150));
+
+        // Clear MPTCanEscrow
+        mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanEscrow});
+        env(escrow::create(carol, bob, MPT(3)),
+            escrow::condition(escrow::cb1),
+            escrow::finish_time(env.now() + 1s),
+            fee(baseFee * 150),
+            ter(tecNO_PERMISSION));
+    }
+
+    void
+    testMutateCanTransfer(FeatureBitset features)
+    {
+        testcase("Mutate MPTCanTransfer");
+
+        using namespace test::jtx;
+        Account const alice("alice");
+        Account const bob("bob");
+        Account const carol("carol");
+
+        {
+            Env env{*this, features};
+
+            MPTTester mptAlice(env, alice, {.holders = {bob, carol}});
+            mptAlice.create(
+                {.ownerCount = 1,
+                 .mutableFlags =
+                     tfMPTCanMutateCanTransfer | tfMPTCanMutateTransferFee});
+
+            mptAlice.authorize({.account = bob});
+            mptAlice.authorize({.account = carol});
+
+            // Pay to bob
+            mptAlice.pay(alice, bob, 1000);
+
+            // Bob can not pay carol since MPTCanTransfer is not set
+            mptAlice.pay(bob, carol, 50, tecNO_AUTH);
+
+            // Can not set non-zero transfer fee when MPTCanTransfer is not set
+            mptAlice.set(
+                {.account = alice,
+                 .transferFee = 100,
+                 .err = tecNO_PERMISSION});
+
+            // Can not set non-zero transfer fee even when trying to set
+            // MPTCanTransfer at the same time
+            mptAlice.set(
+                {.account = alice,
+                 .mutableFlags = tfMPTSetCanTransfer,
+                 .transferFee = 100,
+                 .err = tecNO_PERMISSION});
+
+            // Alice sets MPTCanTransfer
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTSetCanTransfer});
+
+            // Can set transfer fee now
+            BEAST_EXPECT(!mptAlice.isTransferFeePresent());
+            mptAlice.set({.account = alice, .transferFee = 100});
+            BEAST_EXPECT(mptAlice.isTransferFeePresent());
+
+            // Bob can pay carol
+            mptAlice.pay(bob, carol, 50);
+
+            // Alice clears MPTCanTransfer
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTClearCanTransfer});
+
+            // TransferFee field is removed when MPTCanTransfer is cleared
+            BEAST_EXPECT(!mptAlice.isTransferFeePresent());
+
+            // Bob can not pay
+            mptAlice.pay(bob, carol, 50, tecNO_AUTH);
+        }
+
+        // Can set transfer fee to zero when MPTCanTransfer is not set, but
+        // tfMPTCanMutateTransferFee is set.
+        {
+            Env env{*this, features};
+
+            MPTTester mptAlice(env, alice, {.holders = {bob, carol}});
+            mptAlice.create(
+                {.transferFee = 100,
+                 .ownerCount = 1,
+                 .flags = tfMPTCanTransfer,
+                 .mutableFlags =
+                     tfMPTCanMutateTransferFee | tfMPTCanMutateCanTransfer});
+
+            BEAST_EXPECT(mptAlice.checkTransferFee(100));
+
+            // Clear MPTCanTransfer and transfer fee is removed
+            mptAlice.set(
+                {.account = alice, .mutableFlags = tfMPTClearCanTransfer});
+            BEAST_EXPECT(!mptAlice.isTransferFeePresent());
+
+            // Can still set transfer fee to zero, although it is already zero
+            mptAlice.set({.account = alice, .transferFee = 0});
+
+            // TransferFee field is still not present
+            BEAST_EXPECT(!mptAlice.isTransferFeePresent());
+        }
+    }
+
+    void
+    testMutateCanClawback(FeatureBitset features)
+    {
+        testcase("Mutate MPTCanClawback");
+
+        using namespace test::jtx;
+        Env env(*this, features);
+        Account const alice{"alice"};
+        Account const bob{"bob"};
+
+        MPTTester mptAlice(env, alice, {.holders = {bob}});
+
+        mptAlice.create(
+            {.ownerCount = 1,
+             .holderCount = 0,
+             .mutableFlags = tfMPTCanMutateCanClawback});
+
+        // Bob creates an MPToken
+        mptAlice.authorize({.account = bob});
+
+        // Alice pays bob 100 tokens
+        mptAlice.pay(alice, bob, 100);
+
+        // MPTCanClawback is not enabled
+        mptAlice.claw(alice, bob, 1, tecNO_PERMISSION);
+
+        // Enable MPTCanClawback
+        mptAlice.set({.account = alice, .mutableFlags = tfMPTSetCanClawback});
+
+        // Can clawback now
+        mptAlice.claw(alice, bob, 1);
+
+        // Clear MPTCanClawback
+        mptAlice.set({.account = alice, .mutableFlags = tfMPTClearCanClawback});
+
+        // Can not clawback
+        mptAlice.claw(alice, bob, 1, tecNO_PERMISSION);
+    }
+
 public:
     void
     run() override
@@ -2747,39 +3625,39 @@ public:
 
         // MPTokenIssuanceCreate
         testCreateValidation(all - featureSingleAssetVault);
-        testCreateValidation(
-            (all | featureSingleAssetVault) - featurePermissionedDomains);
-        testCreateValidation(all | featureSingleAssetVault);
+        testCreateValidation(all - featurePermissionedDomains);
+        testCreateValidation(all);
         testCreateEnabled(all - featureSingleAssetVault);
-        testCreateEnabled(all | featureSingleAssetVault);
+        testCreateEnabled(all);
 
         // MPTokenIssuanceDestroy
         testDestroyValidation(all - featureSingleAssetVault);
-        testDestroyValidation(all | featureSingleAssetVault);
+        testDestroyValidation(all);
         testDestroyEnabled(all - featureSingleAssetVault);
-        testDestroyEnabled(all | featureSingleAssetVault);
+        testDestroyEnabled(all);
 
         // MPTokenAuthorize
         testAuthorizeValidation(all - featureSingleAssetVault);
-        testAuthorizeValidation(all | featureSingleAssetVault);
+        testAuthorizeValidation(all);
         testAuthorizeEnabled(all - featureSingleAssetVault);
-        testAuthorizeEnabled(all | featureSingleAssetVault);
+        testAuthorizeEnabled(all);
 
         // MPTokenIssuanceSet
+        testSetValidation(all - featureSingleAssetVault - featureDynamicMPT);
         testSetValidation(all - featureSingleAssetVault);
-        testSetValidation(
-            (all | featureSingleAssetVault) - featurePermissionedDomains);
-        testSetValidation(all | featureSingleAssetVault);
+        testSetValidation(all - featureDynamicMPT);
+        testSetValidation(all - featurePermissionedDomains);
+        testSetValidation(all);
 
         testSetEnabled(all - featureSingleAssetVault);
-        testSetEnabled(all | featureSingleAssetVault);
+        testSetEnabled(all);
 
         // MPT clawback
         testClawbackValidation(all);
         testClawback(all);
 
         // Test Direct Payment
-        testPayment(all | featureSingleAssetVault);
+        testPayment(all);
         testDepositPreauth(all);
         testDepositPreauth(all - featureCredentials);
 
@@ -2794,6 +3672,16 @@ public:
 
         // Test helpers
         testHelperFunctions();
+
+        // Dynamic MPT
+        testInvalidCreateDynamic(all);
+        testInvalidSetDynamic(all);
+        testMutateMPT(all);
+        testMutateCanLock(all);
+        testMutateRequireAuth(all);
+        testMutateCanEscrow(all);
+        testMutateCanTransfer(all);
+        testMutateCanClawback(all);
     }
 };
 

src/test/app/NetworkOPs_test.cpp

@@ -0,0 +1,80 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2020 Dev Null Productions
+
+    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.
+*/
+//==============================================================================
+
+#include <test/jtx.h>
+#include <test/jtx/CaptureLogs.h>
+#include <test/jtx/Env.h>
+
+#include <xrpld/app/misc/HashRouter.h>
+
+namespace ripple {
+namespace test {
+
+class NetworkOPs_test : public beast::unit_test::suite
+{
+public:
+    void
+    run() override
+    {
+        testAllBadHeldTransactions();
+    }
+
+    void
+    testAllBadHeldTransactions()
+    {
+        // All trasactions are already marked as SF_BAD, and we should be able
+        // to handle the case properly without an assertion failure
+        testcase("No valid transactions in batch");
+
+        std::string logs;
+
+        {
+            using namespace jtx;
+            auto const alice = Account{"alice"};
+            Env env{
+                *this,
+                envconfig(),
+                std::make_unique<CaptureLogs>(&logs),
+                beast::severities::kAll};
+            env.memoize(env.master);
+            env.memoize(alice);
+
+            auto const jtx = env.jt(ticket::create(alice, 1), seq(1), fee(10));
+
+            auto transacionId = jtx.stx->getTransactionID();
+            env.app().getHashRouter().setFlags(
+                transacionId, HashRouterFlags::HELD);
+
+            env(jtx, json(jss::Sequence, 1), ter(terNO_ACCOUNT));
+
+            env.app().getHashRouter().setFlags(
+                transacionId, HashRouterFlags::BAD);
+
+            env.close();
+        }
+
+        BEAST_EXPECT(
+            logs.find("No transaction to process!") != std::string::npos);
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(NetworkOPs, app, ripple);
+
+}  // namespace test
+}  // namespace ripple

src/test/app/SHAMapStore_test.cpp

@@ -558,23 +558,8 @@ public:
         Env env(*this, envconfig(onlineDelete));
 
         /////////////////////////////////////////////////////////////
-        // Create the backend. Normally, SHAMapStoreImp handles all these
-        // details
-        auto nscfg = env.app().config().section(ConfigSection::nodeDatabase());
-
-        // Provide default values:
-        if (!nscfg.exists("cache_size"))
-            nscfg.set(
-                "cache_size",
-                std::to_string(env.app().config().getValueFor(
-                    SizedItem::treeCacheSize, std::nullopt)));
-
-        if (!nscfg.exists("cache_age"))
-            nscfg.set(
-                "cache_age",
-                std::to_string(env.app().config().getValueFor(
-                    SizedItem::treeCacheAge, std::nullopt)));
-
+        // Create NodeStore with two backends to allow online deletion of data.
+        // Normally, SHAMapStoreImp handles all these details.
         NodeStoreScheduler scheduler(env.app().getJobQueue());
 
         std::string const writableDb = "write";
@@ -582,9 +567,8 @@ public:
         auto writableBackend = makeBackendRotating(env, scheduler, writableDb);
         auto archiveBackend = makeBackendRotating(env, scheduler, archiveDb);
 
-        // Create NodeStore with two backends to allow online deletion of
-        // data
         constexpr int readThreads = 4;
+        auto nscfg = env.app().config().section(ConfigSection::nodeDatabase());
         auto dbr = std::make_unique<NodeStore::DatabaseRotatingImp>(
             scheduler,
             readThreads,

src/test/jtx/impl/mpt.cpp

@@ -102,6 +102,8 @@ MPTTester::create(MPTCreate const& arg)
         jv[sfMaximumAmount] = std::to_string(*arg.maxAmt);
     if (arg.domainID)
         jv[sfDomainID] = to_string(*arg.domainID);
+    if (arg.mutableFlags)
+        jv[sfMutableFlags] = *arg.mutableFlags;
     if (submit(arg, jv) != tesSUCCESS)
     {
         // Verify issuance doesn't exist
@@ -240,19 +242,59 @@ MPTTester::set(MPTSet const& arg)
         jv[sfDelegate] = arg.delegate->human();
     if (arg.domainID)
         jv[sfDomainID] = to_string(*arg.domainID);
-    if (submit(arg, jv) == tesSUCCESS && arg.flags.value_or(0))
+    if (arg.mutableFlags)
+        jv[sfMutableFlags] = *arg.mutableFlags;
+    if (arg.transferFee)
+        jv[sfTransferFee] = *arg.transferFee;
+    if (arg.metadata)
+        jv[sfMPTokenMetadata] = strHex(*arg.metadata);
+    if (submit(arg, jv) == tesSUCCESS && (arg.flags || arg.mutableFlags))
     {
         auto require = [&](std::optional<Account> const& holder,
                            bool unchanged) {
             auto flags = getFlags(holder);
             if (!unchanged)
             {
-                if (*arg.flags & tfMPTLock)
-                    flags |= lsfMPTLocked;
-                else if (*arg.flags & tfMPTUnlock)
-                    flags &= ~lsfMPTLocked;
-                else
-                    Throw<std::runtime_error>("Invalid flags");
+                if (arg.flags)
+                {
+                    if (*arg.flags & tfMPTLock)
+                        flags |= lsfMPTLocked;
+                    else if (*arg.flags & tfMPTUnlock)
+                        flags &= ~lsfMPTLocked;
+                }
+
+                if (arg.mutableFlags)
+                {
+                    if (*arg.mutableFlags & tfMPTSetCanLock)
+                        flags |= lsfMPTCanLock;
+                    else if (*arg.mutableFlags & tfMPTClearCanLock)
+                        flags &= ~lsfMPTCanLock;
+
+                    if (*arg.mutableFlags & tfMPTSetRequireAuth)
+                        flags |= lsfMPTRequireAuth;
+                    else if (*arg.mutableFlags & tfMPTClearRequireAuth)
+                        flags &= ~lsfMPTRequireAuth;
+
+                    if (*arg.mutableFlags & tfMPTSetCanEscrow)
+                        flags |= lsfMPTCanEscrow;
+                    else if (*arg.mutableFlags & tfMPTClearCanEscrow)
+                        flags &= ~lsfMPTCanEscrow;
+
+                    if (*arg.mutableFlags & tfMPTSetCanClawback)
+                        flags |= lsfMPTCanClawback;
+                    else if (*arg.mutableFlags & tfMPTClearCanClawback)
+                        flags &= ~lsfMPTCanClawback;
+
+                    if (*arg.mutableFlags & tfMPTSetCanTrade)
+                        flags |= lsfMPTCanTrade;
+                    else if (*arg.mutableFlags & tfMPTClearCanTrade)
+                        flags &= ~lsfMPTCanTrade;
+
+                    if (*arg.mutableFlags & tfMPTSetCanTransfer)
+                        flags |= lsfMPTCanTransfer;
+                    else if (*arg.mutableFlags & tfMPTClearCanTransfer)
+                        flags &= ~lsfMPTCanTransfer;
+                }
             }
             env_.require(mptflags(*this, flags, holder));
         };
@@ -313,6 +355,43 @@ MPTTester::checkFlags(
     return expectedFlags == getFlags(holder);
 }
 
+[[nodiscard]] bool
+MPTTester::checkMetadata(std::string const& metadata) const
+{
+    return forObject([&](SLEP const& sle) -> bool {
+        if (sle->isFieldPresent(sfMPTokenMetadata))
+            return strHex(sle->getFieldVL(sfMPTokenMetadata)) ==
+                strHex(metadata);
+        return false;
+    });
+}
+
+[[nodiscard]] bool
+MPTTester::isMetadataPresent() const
+{
+    return forObject([&](SLEP const& sle) -> bool {
+        return sle->isFieldPresent(sfMPTokenMetadata);
+    });
+}
+
+[[nodiscard]] bool
+MPTTester::checkTransferFee(std::uint16_t transferFee) const
+{
+    return forObject([&](SLEP const& sle) -> bool {
+        if (sle->isFieldPresent(sfTransferFee))
+            return sle->getFieldU16(sfTransferFee) == transferFee;
+        return false;
+    });
+}
+
+[[nodiscard]] bool
+MPTTester::isTransferFeePresent() const
+{
+    return forObject([&](SLEP const& sle) -> bool {
+        return sle->isFieldPresent(sfTransferFee);
+    });
+}
+
 void
 MPTTester::pay(
     Account const& src,

src/test/jtx/mpt.h

@@ -106,6 +106,7 @@ struct MPTCreate
     std::optional<std::uint32_t> holderCount = std::nullopt;
     bool fund = true;
     std::optional<std::uint32_t> flags = {0};
+    std::optional<std::uint32_t> mutableFlags = std::nullopt;
     std::optional<uint256> domainID = std::nullopt;
     std::optional<TER> err = std::nullopt;
 };
@@ -139,6 +140,9 @@ struct MPTSet
     std::optional<std::uint32_t> ownerCount = std::nullopt;
     std::optional<std::uint32_t> holderCount = std::nullopt;
     std::optional<std::uint32_t> flags = std::nullopt;
+    std::optional<std::uint32_t> mutableFlags = std::nullopt;
+    std::optional<std::uint16_t> transferFee = std::nullopt;
+    std::optional<std::string> metadata = std::nullopt;
     std::optional<Account> delegate = std::nullopt;
     std::optional<uint256> domainID = std::nullopt;
     std::optional<TER> err = std::nullopt;
@@ -182,6 +186,18 @@ public:
         uint32_t const expectedFlags,
         std::optional<Account> const& holder = std::nullopt) const;
 
+    [[nodiscard]] bool
+    checkMetadata(std::string const& metadata) const;
+
+    [[nodiscard]] bool
+    isMetadataPresent() const;
+
+    [[nodiscard]] bool
+    checkTransferFee(std::uint16_t transferFee) const;
+
+    [[nodiscard]] bool
+    isTransferFeePresent() const;
+
     Account const&
     issuer() const
     {

src/xrpld/app/misc/NetworkOPs.cpp

@@ -1452,6 +1452,11 @@ NetworkOPsImp::processTransactionSet(CanonicalTXSet const& set)
         for (auto& t : transactions)
             mTransactions.push_back(std::move(t));
     }
+    if (mTransactions.empty())
+    {
+        JLOG(m_journal.debug()) << "No transaction to process!";
+        return;
+    }
 
     doTransactionSyncBatch(lock, [&](std::unique_lock<std::mutex> const&) {
         XRPL_ASSERT(

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -162,20 +162,6 @@ std::unique_ptr<NodeStore::Database>
 SHAMapStoreImp::makeNodeStore(int readThreads)
 {
     auto nscfg = app_.config().section(ConfigSection::nodeDatabase());
-
-    // Provide default values:
-    if (!nscfg.exists("cache_size"))
-        nscfg.set(
-            "cache_size",
-            std::to_string(app_.config().getValueFor(
-                SizedItem::treeCacheSize, std::nullopt)));
-
-    if (!nscfg.exists("cache_age"))
-        nscfg.set(
-            "cache_age",
-            std::to_string(app_.config().getValueFor(
-                SizedItem::treeCacheAge, std::nullopt)));
-
     std::unique_ptr<NodeStore::Database> db;
 
     if (deleteInterval_)
@@ -269,8 +255,6 @@ SHAMapStoreImp::run()
     LedgerIndex lastRotated = state_db_.getState().lastRotated;
     netOPs_ = &app_.getOPs();
     ledgerMaster_ = &app_.getLedgerMaster();
-    fullBelowCache_ = &(*app_.getNodeFamily().getFullBelowCache());
-    treeNodeCache_ = &(*app_.getNodeFamily().getTreeNodeCache());
 
     if (advisoryDelete_)
         canDelete_ = state_db_.getCanDelete();
@@ -563,16 +547,13 @@ void
 SHAMapStoreImp::clearCaches(LedgerIndex validatedSeq)
 {
     ledgerMaster_->clearLedgerCachePrior(validatedSeq);
-    fullBelowCache_->clear();
 }
 
 void
 SHAMapStoreImp::freshenCaches()
 {
-    if (freshenCache(*treeNodeCache_))
-        return;
-    if (freshenCache(app_.getMasterTransaction().getCache()))
-        return;
+    freshenCache(*app_.getNodeFamily().getTreeNodeCache());
+    freshenCache(app_.getMasterTransaction().getCache());
 }
 
 void

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -112,8 +112,6 @@ private:
     // as of run() or before
     NetworkOPs* netOPs_ = nullptr;
     LedgerMaster* ledgerMaster_ = nullptr;
-    FullBelowCache* fullBelowCache_ = nullptr;
-    TreeNodeCache* treeNodeCache_ = nullptr;
 
     static constexpr auto nodeStoreName_ = "NodeStore";
 

src/xrpld/app/tx/detail/Escrow.cpp

@@ -1007,8 +1007,13 @@ escrowUnlockApplyHelper<MPTIssue>(
         // compute balance to transfer
         finalAmt = amount.value() - xferFee;
     }
-
-    return rippleUnlockEscrowMPT(view, sender, receiver, finalAmt, journal);
+    return rippleUnlockEscrowMPT(
+        view,
+        sender,
+        receiver,
+        finalAmt,
+        view.rules().enabled(fixTokenEscrowV1) ? amount : finalAmt,
+        journal);
 }
 
 TER

src/xrpld/app/tx/detail/MPTokenIssuanceCreate.cpp

@@ -36,9 +36,17 @@ MPTokenIssuanceCreate::preflight(PreflightContext const& ctx)
           ctx.rules.enabled(featureSingleAssetVault)))
         return temDISABLED;
 
+    if (ctx.tx.isFieldPresent(sfMutableFlags) &&
+        !ctx.rules.enabled(featureDynamicMPT))
+        return temDISABLED;
+
     if (auto const ret = preflight1(ctx); !isTesSuccess(ret))
         return ret;
 
+    if (auto const mutableFlags = ctx.tx[~sfMutableFlags]; mutableFlags &&
+        (!*mutableFlags || *mutableFlags & tfMPTokenIssuanceCreateMutableMask))
+        return temINVALID_FLAG;
+
     if (ctx.tx.getFlags() & tfMPTokenIssuanceCreateMask)
         return temINVALID_FLAG;
 
@@ -132,6 +140,9 @@ MPTokenIssuanceCreate::create(
         if (args.domainId)
             (*mptIssuance)[sfDomainID] = *args.domainId;
 
+        if (args.mutableFlags)
+            (*mptIssuance)[sfMutableFlags] = *args.mutableFlags;
+
         view.insert(mptIssuance);
     }
 
@@ -158,6 +169,7 @@ MPTokenIssuanceCreate::doApply()
             .transferFee = tx[~sfTransferFee],
             .metadata = tx[~sfMPTokenMetadata],
             .domainId = tx[~sfDomainID],
+            .mutableFlags = tx[~sfMutableFlags],
         });
     return result ? tesSUCCESS : result.error();
 }

src/xrpld/app/tx/detail/MPTokenIssuanceCreate.h

@@ -38,6 +38,7 @@ struct MPTCreateArgs
     std::optional<std::uint16_t> transferFee{};
     std::optional<Slice> const& metadata{};
     std::optional<uint256> domainId{};
+    std::optional<std::uint32_t> mutableFlags{};
 };
 
 class MPTokenIssuanceCreate : public Transactor

src/xrpld/app/tx/detail/MPTokenIssuanceSet.cpp

@@ -26,6 +26,24 @@
 
 namespace ripple {
 
+// Maps set/clear mutable flags in an MPTokenIssuanceSet transaction to the
+// corresponding ledger mutable flags that control whether the change is
+// allowed.
+struct MPTMutabilityFlags
+{
+    std::uint32_t setFlag;
+    std::uint32_t clearFlag;
+    std::uint32_t canMutateFlag;
+};
+
+static constexpr std::array<MPTMutabilityFlags, 6> mptMutabilityFlags = {
+    {{tfMPTSetCanLock, tfMPTClearCanLock, lsfMPTCanMutateCanLock},
+     {tfMPTSetRequireAuth, tfMPTClearRequireAuth, lsfMPTCanMutateRequireAuth},
+     {tfMPTSetCanEscrow, tfMPTClearCanEscrow, lsfMPTCanMutateCanEscrow},
+     {tfMPTSetCanTrade, tfMPTClearCanTrade, lsfMPTCanMutateCanTrade},
+     {tfMPTSetCanTransfer, tfMPTClearCanTransfer, lsfMPTCanMutateCanTransfer},
+     {tfMPTSetCanClawback, tfMPTClearCanClawback, lsfMPTCanMutateCanClawback}}};
+
 NotTEC
 MPTokenIssuanceSet::preflight(PreflightContext const& ctx)
 {
@@ -37,6 +55,14 @@ MPTokenIssuanceSet::preflight(PreflightContext const& ctx)
           ctx.rules.enabled(featureSingleAssetVault)))
         return temDISABLED;
 
+    auto const mutableFlags = ctx.tx[~sfMutableFlags];
+    auto const metadata = ctx.tx[~sfMPTokenMetadata];
+    auto const transferFee = ctx.tx[~sfTransferFee];
+    auto const isMutate = mutableFlags || metadata || transferFee;
+
+    if (isMutate && !ctx.rules.enabled(featureDynamicMPT))
+        return temDISABLED;
+
     if (ctx.tx.isFieldPresent(sfDomainID) && ctx.tx.isFieldPresent(sfHolder))
         return temMALFORMED;
 
@@ -57,13 +83,54 @@ MPTokenIssuanceSet::preflight(PreflightContext const& ctx)
     if (holderID && accountID == holderID)
         return temMALFORMED;
 
-    if (ctx.rules.enabled(featureSingleAssetVault))
+    if (ctx.rules.enabled(featureSingleAssetVault) ||
+        ctx.rules.enabled(featureDynamicMPT))
     {
         // Is this transaction actually changing anything ?
-        if (txFlags == 0 && !ctx.tx.isFieldPresent(sfDomainID))
+        if (txFlags == 0 && !ctx.tx.isFieldPresent(sfDomainID) && !isMutate)
             return temMALFORMED;
     }
 
+    if (ctx.rules.enabled(featureDynamicMPT))
+    {
+        // Holder field is not allowed when mutating MPTokenIssuance
+        if (isMutate && holderID)
+            return temMALFORMED;
+
+        // Can not set flags when mutating MPTokenIssuance
+        if (isMutate && (txFlags & tfUniversalMask))
+            return temMALFORMED;
+
+        if (transferFee && *transferFee > maxTransferFee)
+            return temBAD_TRANSFER_FEE;
+
+        if (metadata && metadata->length() > maxMPTokenMetadataLength)
+            return temMALFORMED;
+
+        if (mutableFlags)
+        {
+            if (!*mutableFlags ||
+                (*mutableFlags & tfMPTokenIssuanceSetMutableMask))
+                return temINVALID_FLAG;
+
+            // Can not set and clear the same flag
+            if (std::any_of(
+                    mptMutabilityFlags.begin(),
+                    mptMutabilityFlags.end(),
+                    [mutableFlags](auto const& f) {
+                        return (*mutableFlags & f.setFlag) &&
+                            (*mutableFlags & f.clearFlag);
+                    }))
+                return temINVALID_FLAG;
+
+            // Trying to set a non-zero TransferFee and clear MPTCanTransfer
+            // in the same transaction is not allowed.
+            if (transferFee.value_or(0) &&
+                (*mutableFlags & tfMPTClearCanTransfer))
+                return temMALFORMED;
+        }
+    }
+
     return preflight2(ctx);
 }
 
@@ -116,7 +183,8 @@ MPTokenIssuanceSet::preclaim(PreclaimContext const& ctx)
     if (!sleMptIssuance->isFlag(lsfMPTCanLock))
     {
         // For readability two separate `if` rather than `||` of two conditions
-        if (!ctx.view.rules().enabled(featureSingleAssetVault))
+        if (!ctx.view.rules().enabled(featureSingleAssetVault) &&
+            !ctx.view.rules().enabled(featureDynamicMPT))
             return tecNO_PERMISSION;
         else if (ctx.tx.isFlag(tfMPTLock) || ctx.tx.isFlag(tfMPTUnlock))
             return tecNO_PERMISSION;
@@ -152,6 +220,44 @@ MPTokenIssuanceSet::preclaim(PreclaimContext const& ctx)
         }
     }
 
+    // sfMutableFlags is soeDEFAULT, defaulting to 0 if not specified on
+    // the ledger.
+    auto const currentMutableFlags =
+        sleMptIssuance->getFieldU32(sfMutableFlags);
+
+    auto isMutableFlag = [&](std::uint32_t mutableFlag) -> bool {
+        return currentMutableFlags & mutableFlag;
+    };
+
+    if (auto const mutableFlags = ctx.tx[~sfMutableFlags])
+    {
+        if (std::any_of(
+                mptMutabilityFlags.begin(),
+                mptMutabilityFlags.end(),
+                [mutableFlags, &isMutableFlag](auto const& f) {
+                    return !isMutableFlag(f.canMutateFlag) &&
+                        ((*mutableFlags & (f.setFlag | f.clearFlag)));
+                }))
+            return tecNO_PERMISSION;
+    }
+
+    if (!isMutableFlag(lsfMPTCanMutateMetadata) &&
+        ctx.tx.isFieldPresent(sfMPTokenMetadata))
+        return tecNO_PERMISSION;
+
+    if (auto const fee = ctx.tx[~sfTransferFee])
+    {
+        // A non-zero TransferFee is only valid if the lsfMPTCanTransfer flag
+        // was previously enabled (at issuance or via a prior mutation). Setting
+        // it by tfMPTSetCanTransfer in the current transaction does not meet
+        // this requirement.
+        if (fee > 0u && !sleMptIssuance->isFlag(lsfMPTCanTransfer))
+            return tecNO_PERMISSION;
+
+        if (!isMutableFlag(lsfMPTCanMutateTransferFee))
+            return tecNO_PERMISSION;
+    }
+
     return tesSUCCESS;
 }
 
@@ -180,9 +286,47 @@ MPTokenIssuanceSet::doApply()
     else if (txFlags & tfMPTUnlock)
         flagsOut &= ~lsfMPTLocked;
 
+    if (auto const mutableFlags = ctx_.tx[~sfMutableFlags].value_or(0))
+    {
+        for (auto const& f : mptMutabilityFlags)
+        {
+            if (mutableFlags & f.setFlag)
+                flagsOut |= f.canMutateFlag;
+            else if (mutableFlags & f.clearFlag)
+                flagsOut &= ~f.canMutateFlag;
+        }
+
+        if (mutableFlags & tfMPTClearCanTransfer)
+        {
+            // If the lsfMPTCanTransfer flag is being cleared, then also clear
+            // the TransferFee field.
+            sle->makeFieldAbsent(sfTransferFee);
+        }
+    }
+
     if (flagsIn != flagsOut)
         sle->setFieldU32(sfFlags, flagsOut);
 
+    if (auto const transferFee = ctx_.tx[~sfTransferFee])
+    {
+        // TransferFee uses soeDEFAULT style:
+        // - If the field is absent, it is interpreted as 0.
+        // - If the field is present, it must be non-zero.
+        // Therefore, when TransferFee is 0, the field should be removed.
+        if (transferFee == 0)
+            sle->makeFieldAbsent(sfTransferFee);
+        else
+            sle->setFieldU16(sfTransferFee, *transferFee);
+    }
+
+    if (auto const metadata = ctx_.tx[~sfMPTokenMetadata])
+    {
+        if (metadata->empty())
+            sle->makeFieldAbsent(sfMPTokenMetadata);
+        else
+            sle->setFieldVL(sfMPTokenMetadata, *metadata);
+    }
+
     if (domainID)
     {
         // This is enforced in preflight.

src/xrpld/ledger/View.h

@@ -719,7 +719,8 @@ rippleUnlockEscrowMPT(
     ApplyView& view,
     AccountID const& uGrantorID,
     AccountID const& uGranteeID,
-    STAmount const& saAmount,
+    STAmount const& netAmount,
+    STAmount const& grossAmount,
     beast::Journal j);
 
 /** Calls static accountSendIOU if saAmount represents Issue.

src/xrpld/ledger/detail/View.cpp

@@ -3006,11 +3006,17 @@ rippleUnlockEscrowMPT(
     ApplyView& view,
     AccountID const& sender,
     AccountID const& receiver,
-    STAmount const& amount,
+    STAmount const& netAmount,
+    STAmount const& grossAmount,
     beast::Journal j)
 {
-    auto const issuer = amount.getIssuer();
-    auto const mptIssue = amount.get<MPTIssue>();
+    if (!view.rules().enabled(fixTokenEscrowV1))
+        XRPL_ASSERT(
+            netAmount == grossAmount,
+            "ripple::rippleUnlockEscrowMPT : netAmount == grossAmount");
+
+    auto const& issuer = netAmount.getIssuer();
+    auto const& mptIssue = netAmount.get<MPTIssue>();
     auto const mptID = keylet::mptIssuance(mptIssue.getMptID());
     auto sleIssuance = view.peek(mptID);
     if (!sleIssuance)
@@ -3031,7 +3037,7 @@ rippleUnlockEscrowMPT(
         }  // LCOV_EXCL_STOP
 
         auto const locked = sleIssuance->getFieldU64(sfLockedAmount);
-        auto const redeem = amount.mpt().value();
+        auto const redeem = grossAmount.mpt().value();
 
         // Underflow check for subtraction
         if (!canSubtract(
@@ -3064,7 +3070,7 @@ rippleUnlockEscrowMPT(
         }  // LCOV_EXCL_STOP
 
         auto current = sle->getFieldU64(sfMPTAmount);
-        auto delta = amount.mpt().value();
+        auto delta = netAmount.mpt().value();
 
         // Overflow check for addition
         if (!canAdd(STAmount(mptIssue, current), STAmount(mptIssue, delta)))
@@ -3082,7 +3088,7 @@ rippleUnlockEscrowMPT(
     {
         // Decrease the Issuance OutstandingAmount
         auto const outstanding = sleIssuance->getFieldU64(sfOutstandingAmount);
-        auto const redeem = amount.mpt().value();
+        auto const redeem = netAmount.mpt().value();
 
         // Underflow check for subtraction
         if (!canSubtract(
@@ -3126,7 +3132,7 @@ rippleUnlockEscrowMPT(
         }  // LCOV_EXCL_STOP
 
         auto const locked = sle->getFieldU64(sfLockedAmount);
-        auto const delta = amount.mpt().value();
+        auto const delta = grossAmount.mpt().value();
 
         // Underflow check for subtraction
         if (!canSubtract(STAmount(mptIssue, locked), STAmount(mptIssue, delta)))
@@ -3144,6 +3150,28 @@ rippleUnlockEscrowMPT(
             sle->setFieldU64(sfLockedAmount, newLocked);
         view.update(sle);
     }
+
+    // Note: The gross amount is the amount that was locked, the net
+    // amount is the amount that is being unlocked. The difference is the fee
+    // that was charged for the transfer. If this difference is greater than
+    // zero, we need to update the outstanding amount.
+    auto const diff = grossAmount.mpt().value() - netAmount.mpt().value();
+    if (diff != 0)
+    {
+        auto const outstanding = sleIssuance->getFieldU64(sfOutstandingAmount);
+        // Underflow check for subtraction
+        if (!canSubtract(
+                STAmount(mptIssue, outstanding), STAmount(mptIssue, diff)))
+        {  // LCOV_EXCL_START
+            JLOG(j.error())
+                << "rippleUnlockEscrowMPT: insufficient outstanding amount for "
+                << mptIssue.getMptID() << ": " << outstanding << " < " << diff;
+            return tecINTERNAL;
+        }  // LCOV_EXCL_STOP
+
+        sleIssuance->setFieldU64(sfOutstandingAmount, outstanding - diff);
+        view.update(sleIssuance);
+    }
     return tesSUCCESS;
 }
 

src/xrpld/nodestore/detail/DatabaseNodeImp.cpp

@@ -33,14 +33,6 @@ DatabaseNodeImp::store(
 
     auto obj = NodeObject::createObject(type, std::move(data), hash);
     backend_->store(obj);
-    if (cache_)
-    {
-        // After the store, replace a negative cache entry if there is one
-        cache_->canonicalize(
-            hash, obj, [](std::shared_ptr<NodeObject> const& n) {
-                return n->getType() == hotDUMMY;
-            });
-    }
 }
 
 void
@@ -49,23 +41,12 @@ DatabaseNodeImp::asyncFetch(
     std::uint32_t ledgerSeq,
     std::function<void(std::shared_ptr<NodeObject> const&)>&& callback)
 {
-    if (cache_)
-    {
-        std::shared_ptr<NodeObject> obj = cache_->fetch(hash);
-        if (obj)
-        {
-            callback(obj->getType() == hotDUMMY ? nullptr : obj);
-            return;
-        }
-    }
     Database::asyncFetch(hash, ledgerSeq, std::move(callback));
 }
 
 void
 DatabaseNodeImp::sweep()
 {
-    if (cache_)
-        cache_->sweep();
 }
 
 std::shared_ptr<NodeObject>
@@ -75,64 +56,33 @@ DatabaseNodeImp::fetchNodeObject(
     FetchReport& fetchReport,
     bool duplicate)
 {
-    std::shared_ptr<NodeObject> nodeObject =
-        cache_ ? cache_->fetch(hash) : nullptr;
+    std::shared_ptr<NodeObject> nodeObject = nullptr;
+    Status status;
 
-    if (!nodeObject)
+    try
     {
-        JLOG(j_.trace()) << "fetchNodeObject " << hash << ": record not "
-                         << (cache_ ? "cached" : "found");
-
-        Status status;
-
-        try
-        {
-            status = backend_->fetch(hash.data(), &nodeObject);
-        }
-        catch (std::exception const& e)
-        {
-            JLOG(j_.fatal())
-                << "fetchNodeObject " << hash
-                << ": Exception fetching from backend: " << e.what();
-            Rethrow();
-        }
-
-        switch (status)
-        {
-            case ok:
-                if (cache_)
-                {
-                    if (nodeObject)
-                        cache_->canonicalize_replace_client(hash, nodeObject);
-                    else
-                    {
-                        auto notFound =
-                            NodeObject::createObject(hotDUMMY, {}, hash);
-                        cache_->canonicalize_replace_client(hash, notFound);
-                        if (notFound->getType() != hotDUMMY)
-                            nodeObject = notFound;
-                    }
-                }
-                break;
-            case notFound:
-                break;
-            case dataCorrupt:
-                JLOG(j_.fatal()) << "fetchNodeObject " << hash
-                                 << ": nodestore data is corrupted";
-                break;
-            default:
-                JLOG(j_.warn())
-                    << "fetchNodeObject " << hash
-                    << ": backend returns unknown result " << status;
-                break;
-        }
+        status = backend_->fetch(hash.data(), &nodeObject);
     }
-    else
+    catch (std::exception const& e)
     {
-        JLOG(j_.trace()) << "fetchNodeObject " << hash
-                         << ": record found in cache";
-        if (nodeObject->getType() == hotDUMMY)
-            nodeObject.reset();
+        JLOG(j_.fatal()) << "fetchNodeObject " << hash
+                         << ": Exception fetching from backend: " << e.what();
+        Rethrow();
+    }
+
+    switch (status)
+    {
+        case ok:
+        case notFound:
+            break;
+        case dataCorrupt:
+            JLOG(j_.fatal()) << "fetchNodeObject " << hash
+                             << ": nodestore data is corrupted";
+            break;
+        default:
+            JLOG(j_.warn()) << "fetchNodeObject " << hash
+                            << ": backend returns unknown result " << status;
+            break;
     }
 
     if (nodeObject)
@@ -144,71 +94,33 @@ DatabaseNodeImp::fetchNodeObject(
 std::vector<std::shared_ptr<NodeObject>>
 DatabaseNodeImp::fetchBatch(std::vector<uint256> const& hashes)
 {
-    std::vector<std::shared_ptr<NodeObject>> results{hashes.size()};
     using namespace std::chrono;
     auto const before = steady_clock::now();
-    std::unordered_map<uint256 const*, size_t> indexMap;
-    std::vector<uint256 const*> cacheMisses;
-    uint64_t hits = 0;
-    uint64_t fetches = 0;
+
+    std::vector<uint256 const*> batch{hashes.size()};
     for (size_t i = 0; i < hashes.size(); ++i)
     {
         auto const& hash = hashes[i];
-        // See if the object already exists in the cache
-        auto nObj = cache_ ? cache_->fetch(hash) : nullptr;
-        ++fetches;
-        if (!nObj)
-        {
-            // Try the database
-            indexMap[&hash] = i;
-            cacheMisses.push_back(&hash);
-        }
-        else
-        {
-            results[i] = nObj->getType() == hotDUMMY ? nullptr : nObj;
-            // It was in the cache.
-            ++hits;
-        }
+        batch.push_back(&hash);
     }
 
-    JLOG(j_.debug()) << "fetchBatch - cache hits = "
-                     << (hashes.size() - cacheMisses.size())
-                     << " - cache misses = " << cacheMisses.size();
-    auto dbResults = backend_->fetchBatch(cacheMisses).first;
-
-    for (size_t i = 0; i < dbResults.size(); ++i)
+    std::vector<std::shared_ptr<NodeObject>> results{hashes.size()};
+    results = backend_->fetchBatch(batch).first;
+    for (size_t i = 0; i < results.size(); ++i)
     {
-        auto nObj = std::move(dbResults[i]);
-        size_t index = indexMap[cacheMisses[i]];
-        auto const& hash = hashes[index];
-
-        if (nObj)
-        {
-            // Ensure all threads get the same object
-            if (cache_)
-                cache_->canonicalize_replace_client(hash, nObj);
-        }
-        else
+        if (!results[i])
         {
             JLOG(j_.error())
                 << "fetchBatch - "
-                << "record not found in db or cache. hash = " << strHex(hash);
-            if (cache_)
-            {
-                auto notFound = NodeObject::createObject(hotDUMMY, {}, hash);
-                cache_->canonicalize_replace_client(hash, notFound);
-                if (notFound->getType() != hotDUMMY)
-                    nObj = std::move(notFound);
-            }
+                << "record not found in db. hash = " << strHex(hashes[i]);
         }
-        results[index] = std::move(nObj);
     }
 
     auto fetchDurationUs =
         std::chrono::duration_cast<std::chrono::microseconds>(
             steady_clock::now() - before)
             .count();
-    updateFetchMetrics(fetches, hits, fetchDurationUs);
+    updateFetchMetrics(hashes.size(), 0, fetchDurationUs);
     return results;
 }
 

src/xrpld/nodestore/detail/DatabaseNodeImp.h

@@ -45,38 +45,6 @@ public:
         : Database(scheduler, readThreads, config, j)
         , backend_(std::move(backend))
     {
-        std::optional<int> cacheSize, cacheAge;
-
-        if (config.exists("cache_size"))
-        {
-            cacheSize = get<int>(config, "cache_size");
-            if (cacheSize.value() < 0)
-            {
-                Throw<std::runtime_error>(
-                    "Specified negative value for cache_size");
-            }
-        }
-
-        if (config.exists("cache_age"))
-        {
-            cacheAge = get<int>(config, "cache_age");
-            if (cacheAge.value() < 0)
-            {
-                Throw<std::runtime_error>(
-                    "Specified negative value for cache_age");
-            }
-        }
-
-        if (cacheSize != 0 || cacheAge != 0)
-        {
-            cache_ = std::make_shared<TaggedCache<uint256, NodeObject>>(
-                "DatabaseNodeImp",
-                cacheSize.value_or(0),
-                std::chrono::minutes(cacheAge.value_or(0)),
-                stopwatch(),
-                j);
-        }
-
         XRPL_ASSERT(
             backend_,
             "ripple::NodeStore::DatabaseNodeImp::DatabaseNodeImp : non-null "
@@ -137,9 +105,6 @@ public:
     sweep() override;
 
 private:
-    // Cache for database objects. This cache is not always initialized. Check
-    // for null before using.
-    std::shared_ptr<TaggedCache<uint256, NodeObject>> cache_;
     // Persistent key/value storage
     std::shared_ptr<Backend> backend_;
 

src/xrpld/overlay/detail/ConnectAttempt.cpp

@@ -24,6 +24,8 @@
 
 #include <xrpl/json/json_reader.h>
 
+#include <sstream>
+
 namespace ripple {
 
 ConnectAttempt::ConnectAttempt(
@@ -45,6 +47,7 @@ ConnectAttempt::ConnectAttempt(
     , usage_(usage)
     , strand_(boost::asio::make_strand(io_context))
     , timer_(io_context)
+    , stepTimer_(io_context)
     , stream_ptr_(std::make_unique<stream_type>(
           socket_type(std::forward<boost::asio::io_context&>(io_context)),
           *context))
@@ -52,14 +55,14 @@ ConnectAttempt::ConnectAttempt(
     , stream_(*stream_ptr_)
     , slot_(slot)
 {
-    JLOG(journal_.debug()) << "Connect " << remote_endpoint;
 }
 
 ConnectAttempt::~ConnectAttempt()
 {
+    // slot_ will be null if we successfully connected
+    // and transferred ownership to a PeerImp
     if (slot_ != nullptr)
         overlay_.peerFinder().on_closed(slot_);
-    JLOG(journal_.trace()) << "~ConnectAttempt";
 }
 
 void
@@ -68,16 +71,29 @@ ConnectAttempt::stop()
     if (!strand_.running_in_this_thread())
         return boost::asio::post(
             strand_, std::bind(&ConnectAttempt::stop, shared_from_this()));
-    if (socket_.is_open())
-    {
-        JLOG(journal_.debug()) << "Stop";
-    }
-    close();
+
+    if (!socket_.is_open())
+        return;
+
+    JLOG(journal_.debug()) << "stop: Stop";
+
+    shutdown();
 }
 
 void
 ConnectAttempt::run()
 {
+    if (!strand_.running_in_this_thread())
+        return boost::asio::post(
+            strand_, std::bind(&ConnectAttempt::run, shared_from_this()));
+
+    JLOG(journal_.debug()) << "run: connecting to " << remote_endpoint_;
+
+    ioPending_ = true;
+
+    // Allow up to connectTimeout_ seconds to establish remote peer connection
+    setTimer(ConnectionStep::TcpConnect);
+
     stream_.next_layer().async_connect(
         remote_endpoint_,
         boost::asio::bind_executor(
@@ -90,61 +106,177 @@ ConnectAttempt::run()
 
 //------------------------------------------------------------------------------
 
+void
+ConnectAttempt::shutdown()
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::ConnectAttempt::shutdown: strand in this thread");
+
+    if (!socket_.is_open())
+        return;
+
+    shutdown_ = true;
+    boost::beast::get_lowest_layer(stream_).cancel();
+
+    tryAsyncShutdown();
+}
+
+void
+ConnectAttempt::tryAsyncShutdown()
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::ConnectAttempt::tryAsyncShutdown : strand in this thread");
+
+    if (!shutdown_ || currentStep_ == ConnectionStep::ShutdownStarted)
+        return;
+
+    if (ioPending_)
+        return;
+
+    // gracefully shutdown the SSL socket, performing a shutdown handshake
+    if (currentStep_ != ConnectionStep::TcpConnect &&
+        currentStep_ != ConnectionStep::TlsHandshake)
+    {
+        setTimer(ConnectionStep::ShutdownStarted);
+        return stream_.async_shutdown(bind_executor(
+            strand_,
+            std::bind(
+                &ConnectAttempt::onShutdown,
+                shared_from_this(),
+                std::placeholders::_1)));
+    }
+
+    close();
+}
+
+void
+ConnectAttempt::onShutdown(error_code ec)
+{
+    cancelTimer();
+
+    if (ec)
+    {
+        // - eof: the stream was cleanly closed
+        // - operation_aborted: an expired timer (slow shutdown)
+        // - stream_truncated: the tcp connection closed (no handshake) it could
+        // occur if a peer does not perform a graceful disconnect
+        // - broken_pipe: the peer is gone
+        // - application data after close notify: benign SSL shutdown condition
+        bool shouldLog =
+            (ec != boost::asio::error::eof &&
+             ec != boost::asio::error::operation_aborted &&
+             ec.message().find("application data after close notify") ==
+                 std::string::npos);
+
+        if (shouldLog)
+        {
+            JLOG(journal_.debug()) << "onShutdown: " << ec.message();
+        }
+    }
+
+    close();
+}
+
 void
 ConnectAttempt::close()
 {
     XRPL_ASSERT(
         strand_.running_in_this_thread(),
         "ripple::ConnectAttempt::close : strand in this thread");
-    if (socket_.is_open())
-    {
-        try
-        {
-            timer_.cancel();
-            socket_.close();
-        }
-        catch (boost::system::system_error const&)
-        {
-            // ignored
-        }
+    if (!socket_.is_open())
+        return;
 
-        JLOG(journal_.debug()) << "Closed";
-    }
+    cancelTimer();
+
+    error_code ec;
+    socket_.close(ec);
 }
 
 void
 ConnectAttempt::fail(std::string const& reason)
 {
     JLOG(journal_.debug()) << reason;
-    close();
+    shutdown();
 }
 
 void
 ConnectAttempt::fail(std::string const& name, error_code ec)
 {
     JLOG(journal_.debug()) << name << ": " << ec.message();
-    close();
+    shutdown();
 }
 
 void
-ConnectAttempt::setTimer()
+ConnectAttempt::setTimer(ConnectionStep step)
 {
-    try
+    currentStep_ = step;
+
+    // Set global timer (only if not already set)
+    if (timer_.expiry() == std::chrono::steady_clock::time_point{})
     {
-        timer_.expires_after(std::chrono::seconds(15));
-    }
-    catch (boost::system::system_error const& e)
-    {
-        JLOG(journal_.error()) << "setTimer: " << e.code();
-        return;
+        try
+        {
+            timer_.expires_after(connectTimeout);
+            timer_.async_wait(boost::asio::bind_executor(
+                strand_,
+                std::bind(
+                    &ConnectAttempt::onTimer,
+                    shared_from_this(),
+                    std::placeholders::_1)));
+        }
+        catch (std::exception const& ex)
+        {
+            JLOG(journal_.error()) << "setTimer (global): " << ex.what();
+            return close();
+        }
     }
 
-    timer_.async_wait(boost::asio::bind_executor(
-        strand_,
-        std::bind(
-            &ConnectAttempt::onTimer,
-            shared_from_this(),
-            std::placeholders::_1)));
+    // Set step-specific timer
+    try
+    {
+        std::chrono::seconds stepTimeout;
+        switch (step)
+        {
+            case ConnectionStep::TcpConnect:
+                stepTimeout = StepTimeouts::tcpConnect;
+                break;
+            case ConnectionStep::TlsHandshake:
+                stepTimeout = StepTimeouts::tlsHandshake;
+                break;
+            case ConnectionStep::HttpWrite:
+                stepTimeout = StepTimeouts::httpWrite;
+                break;
+            case ConnectionStep::HttpRead:
+                stepTimeout = StepTimeouts::httpRead;
+                break;
+            case ConnectionStep::ShutdownStarted:
+                stepTimeout = StepTimeouts::tlsShutdown;
+                break;
+            case ConnectionStep::Complete:
+            case ConnectionStep::Init:
+                return;  // No timer needed for init or complete step
+        }
+
+        // call to expires_after cancels previous timer
+        stepTimer_.expires_after(stepTimeout);
+        stepTimer_.async_wait(boost::asio::bind_executor(
+            strand_,
+            std::bind(
+                &ConnectAttempt::onTimer,
+                shared_from_this(),
+                std::placeholders::_1)));
+
+        JLOG(journal_.trace()) << "setTimer: " << stepToString(step)
+                               << " timeout=" << stepTimeout.count() << "s";
+    }
+    catch (std::exception const& ex)
+    {
+        JLOG(journal_.error())
+            << "setTimer (step " << stepToString(step) << "): " << ex.what();
+        return close();
+    }
 }
 
 void
@@ -153,6 +285,7 @@ ConnectAttempt::cancelTimer()
     try
     {
         timer_.cancel();
+        stepTimer_.cancel();
     }
     catch (boost::system::system_error const&)
     {
@@ -165,34 +298,69 @@ ConnectAttempt::onTimer(error_code ec)
 {
     if (!socket_.is_open())
         return;
-    if (ec == boost::asio::error::operation_aborted)
-        return;
+
     if (ec)
     {
+        // do not initiate shutdown, timers are frequently cancelled
+        if (ec == boost::asio::error::operation_aborted)
+            return;
+
         // This should never happen
         JLOG(journal_.error()) << "onTimer: " << ec.message();
         return close();
     }
-    fail("Timeout");
+
+    // Determine which timer expired by checking their expiry times
+    auto const now = std::chrono::steady_clock::now();
+    bool globalExpired = (timer_.expiry() <= now);
+    bool stepExpired = (stepTimer_.expiry() <= now);
+
+    if (globalExpired)
+    {
+        JLOG(journal_.debug())
+            << "onTimer: Global timeout; step: " << stepToString(currentStep_);
+    }
+    else if (stepExpired)
+    {
+        JLOG(journal_.debug())
+            << "onTimer: Step timeout; step: " << stepToString(currentStep_);
+    }
+    else
+    {
+        JLOG(journal_.warn()) << "onTimer: Unexpected timer callback";
+    }
+
+    close();
 }
 
 void
 ConnectAttempt::onConnect(error_code ec)
 {
-    cancelTimer();
+    ioPending_ = false;
 
-    if (ec == boost::asio::error::operation_aborted)
-        return;
-    endpoint_type local_endpoint;
-    if (!ec)
-        local_endpoint = socket_.local_endpoint(ec);
     if (ec)
+    {
+        if (ec == boost::asio::error::operation_aborted)
+            return tryAsyncShutdown();
+
         return fail("onConnect", ec);
+    }
+
     if (!socket_.is_open())
         return;
-    JLOG(journal_.trace()) << "onConnect";
 
-    setTimer();
+    // check if connection has really been established
+    socket_.local_endpoint(ec);
+    if (ec)
+        return fail("onConnect", ec);
+
+    if (shutdown_)
+        return tryAsyncShutdown();
+
+    ioPending_ = true;
+
+    setTimer(ConnectionStep::TlsHandshake);
+
     stream_.set_verify_mode(boost::asio::ssl::verify_none);
     stream_.async_handshake(
         boost::asio::ssl::stream_base::client,
@@ -207,25 +375,30 @@ ConnectAttempt::onConnect(error_code ec)
 void
 ConnectAttempt::onHandshake(error_code ec)
 {
-    cancelTimer();
-    if (!socket_.is_open())
-        return;
-    if (ec == boost::asio::error::operation_aborted)
-        return;
-    endpoint_type local_endpoint;
-    if (!ec)
-        local_endpoint = socket_.local_endpoint(ec);
+    ioPending_ = false;
+
+    if (ec)
+    {
+        if (ec == boost::asio::error::operation_aborted)
+            return tryAsyncShutdown();
+
+        return fail("onHandshake", ec);
+    }
+
+    auto const local_endpoint = socket_.local_endpoint(ec);
     if (ec)
         return fail("onHandshake", ec);
-    JLOG(journal_.trace()) << "onHandshake";
 
+    setTimer(ConnectionStep::HttpWrite);
+
+    // check if we connected to ourselves
     if (!overlay_.peerFinder().onConnected(
             slot_, beast::IPAddressConversion::from_asio(local_endpoint)))
-        return fail("Duplicate connection");
+        return fail("Self connection");
 
     auto const sharedValue = makeSharedValue(*stream_ptr_, journal_);
     if (!sharedValue)
-        return close();  // makeSharedValue logs
+        return shutdown();  // makeSharedValue logs
 
     req_ = makeRequest(
         !overlay_.peerFinder().config().peerPrivate,
@@ -242,7 +415,11 @@ ConnectAttempt::onHandshake(error_code ec)
         remote_endpoint_.address(),
         app_);
 
-    setTimer();
+    if (shutdown_)
+        return tryAsyncShutdown();
+
+    ioPending_ = true;
+
     boost::beast::http::async_write(
         stream_,
         req_,
@@ -257,13 +434,23 @@ ConnectAttempt::onHandshake(error_code ec)
 void
 ConnectAttempt::onWrite(error_code ec)
 {
-    cancelTimer();
-    if (!socket_.is_open())
-        return;
-    if (ec == boost::asio::error::operation_aborted)
-        return;
+    ioPending_ = false;
+
     if (ec)
+    {
+        if (ec == boost::asio::error::operation_aborted)
+            return tryAsyncShutdown();
+
         return fail("onWrite", ec);
+    }
+
+    if (shutdown_)
+        return tryAsyncShutdown();
+
+    ioPending_ = true;
+
+    setTimer(ConnectionStep::HttpRead);
+
     boost::beast::http::async_read(
         stream_,
         read_buf_,
@@ -280,39 +467,27 @@ void
 ConnectAttempt::onRead(error_code ec)
 {
     cancelTimer();
+    ioPending_ = false;
+    currentStep_ = ConnectionStep::Complete;
 
-    if (!socket_.is_open())
-        return;
-    if (ec == boost::asio::error::operation_aborted)
-        return;
-    if (ec == boost::asio::error::eof)
-    {
-        JLOG(journal_.info()) << "EOF";
-        setTimer();
-        return stream_.async_shutdown(boost::asio::bind_executor(
-            strand_,
-            std::bind(
-                &ConnectAttempt::onShutdown,
-                shared_from_this(),
-                std::placeholders::_1)));
-    }
     if (ec)
-        return fail("onRead", ec);
-    processResponse();
-}
-
-void
-ConnectAttempt::onShutdown(error_code ec)
-{
-    cancelTimer();
-    if (!ec)
     {
-        JLOG(journal_.error()) << "onShutdown: expected error condition";
-        return close();
+        if (ec == boost::asio::error::eof)
+        {
+            JLOG(journal_.debug()) << "EOF";
+            return shutdown();
+        }
+
+        if (ec == boost::asio::error::operation_aborted)
+            return tryAsyncShutdown();
+
+        return fail("onRead", ec);
     }
-    if (ec != boost::asio::error::eof)
-        return fail("onShutdown", ec);
-    close();
+
+    if (shutdown_)
+        return tryAsyncShutdown();
+
+    processResponse();
 }
 
 //--------------------------------------------------------------------------
@@ -320,48 +495,69 @@ ConnectAttempt::onShutdown(error_code ec)
 void
 ConnectAttempt::processResponse()
 {
-    if (response_.result() == boost::beast::http::status::service_unavailable)
-    {
-        Json::Value json;
-        Json::Reader r;
-        std::string s;
-        s.reserve(boost::asio::buffer_size(response_.body().data()));
-        for (auto const buffer : response_.body().data())
-            s.append(
-                static_cast<char const*>(buffer.data()),
-                boost::asio::buffer_size(buffer));
-        auto const success = r.parse(s, json);
-        if (success)
-        {
-            if (json.isObject() && json.isMember("peer-ips"))
-            {
-                Json::Value const& ips = json["peer-ips"];
-                if (ips.isArray())
-                {
-                    std::vector<boost::asio::ip::tcp::endpoint> eps;
-                    eps.reserve(ips.size());
-                    for (auto const& v : ips)
-                    {
-                        if (v.isString())
-                        {
-                            error_code ec;
-                            auto const ep = parse_endpoint(v.asString(), ec);
-                            if (!ec)
-                                eps.push_back(ep);
-                        }
-                    }
-                    overlay_.peerFinder().onRedirects(remote_endpoint_, eps);
-                }
-            }
-        }
-    }
-
     if (!OverlayImpl::isPeerUpgrade(response_))
     {
-        JLOG(journal_.info())
-            << "Unable to upgrade to peer protocol: " << response_.result()
-            << " (" << response_.reason() << ")";
-        return close();
+        // A peer may respond with service_unavailable and a list of alternative
+        // peers to connect to, a differing status code is unexpected
+        if (response_.result() !=
+            boost::beast::http::status::service_unavailable)
+        {
+            JLOG(journal_.warn())
+                << "Unable to upgrade to peer protocol: " << response_.result()
+                << " (" << response_.reason() << ")";
+            return shutdown();
+        }
+
+        // Parse response body to determine if this is a redirect or other
+        // service unavailable
+        std::string responseBody;
+        responseBody.reserve(boost::asio::buffer_size(response_.body().data()));
+        for (auto const buffer : response_.body().data())
+            responseBody.append(
+                static_cast<char const*>(buffer.data()),
+                boost::asio::buffer_size(buffer));
+
+        Json::Value json;
+        Json::Reader reader;
+        auto const isValidJson = reader.parse(responseBody, json);
+
+        // Check if this is a redirect response (contains peer-ips field)
+        auto const isRedirect =
+            isValidJson && json.isObject() && json.isMember("peer-ips");
+
+        if (!isRedirect)
+        {
+            JLOG(journal_.warn())
+                << "processResponse: " << remote_endpoint_
+                << " failed to upgrade to peer protocol: " << response_.result()
+                << " (" << response_.reason() << ")";
+
+            return shutdown();
+        }
+
+        Json::Value const& peerIps = json["peer-ips"];
+        if (!peerIps.isArray())
+            return fail("processResponse: invalid peer-ips format");
+
+        // Extract and validate peer endpoints
+        std::vector<boost::asio::ip::tcp::endpoint> redirectEndpoints;
+        redirectEndpoints.reserve(peerIps.size());
+
+        for (auto const& ipValue : peerIps)
+        {
+            if (!ipValue.isString())
+                continue;
+
+            error_code ec;
+            auto const endpoint = parse_endpoint(ipValue.asString(), ec);
+            if (!ec)
+                redirectEndpoints.push_back(endpoint);
+        }
+
+        // Notify PeerFinder about the redirect redirectEndpoints may be empty
+        overlay_.peerFinder().onRedirects(remote_endpoint_, redirectEndpoints);
+
+        return fail("processResponse: failed to connect to peer: redirected");
     }
 
     // Just because our peer selected a particular protocol version doesn't
@@ -381,11 +577,11 @@ ConnectAttempt::processResponse()
 
     auto const sharedValue = makeSharedValue(*stream_ptr_, journal_);
     if (!sharedValue)
-        return close();  // makeSharedValue logs
+        return shutdown();  // makeSharedValue logs
 
     try
     {
-        auto publicKey = verifyHandshake(
+        auto const publicKey = verifyHandshake(
             response_,
             *sharedValue,
             overlay_.setup().networkID,
@@ -393,11 +589,10 @@ ConnectAttempt::processResponse()
             remote_endpoint_.address(),
             app_);
 
-        JLOG(journal_.info())
-            << "Public Key: " << toBase58(TokenType::NodePublic, publicKey);
-
         JLOG(journal_.debug())
             << "Protocol: " << to_string(*negotiatedProtocol);
+        JLOG(journal_.info())
+            << "Public Key: " << toBase58(TokenType::NodePublic, publicKey);
 
         auto const member = app_.cluster().member(publicKey);
         if (member)
@@ -405,10 +600,21 @@ ConnectAttempt::processResponse()
             JLOG(journal_.info()) << "Cluster name: " << *member;
         }
 
-        auto const result = overlay_.peerFinder().activate(
-            slot_, publicKey, static_cast<bool>(member));
+        auto const result =
+            overlay_.peerFinder().activate(slot_, publicKey, !member->empty());
         if (result != PeerFinder::Result::success)
-            return fail("Outbound " + std::string(to_string(result)));
+        {
+            std::stringstream ss;
+            ss << "Outbound Connect Attempt " << remote_endpoint_ << " "
+               << to_string(result);
+            return fail(ss.str());
+        }
+
+        if (!socket_.is_open())
+            return;
+
+        if (shutdown_)
+            return tryAsyncShutdown();
 
         auto const peer = std::make_shared<PeerImp>(
             app_,

src/xrpld/overlay/detail/ConnectAttempt.h

@@ -22,90 +22,258 @@
 
 #include <xrpld/overlay/detail/OverlayImpl.h>
 
+#include <chrono>
+
 namespace ripple {
 
-/** Manages an outbound connection attempt. */
+/**
+ * @class ConnectAttempt
+ * @brief Manages outbound peer connection attempts with comprehensive timeout
+ * handling
+ *
+ * The ConnectAttempt class handles the complete lifecycle of establishing an
+ * outbound connection to a peer in the XRPL network. It implements a
+ * sophisticated dual-timer system that provides both global timeout protection
+ * and per-step timeout diagnostics.
+ *
+ * The connection establishment follows these steps:
+ * 1. **TCP Connect**: Establish basic network connection
+ * 2. **TLS Handshake**: Negotiate SSL/TLS encryption
+ * 3. **HTTP Write**: Send peer handshake request
+ * 4. **HTTP Read**: Receive and validate peer response
+ * 5. **Complete**: Connection successfully established
+ *
+ * Uses a hybrid timeout approach:
+ * - **Global Timer**: Hard limit (20s) for entire connection process
+ * - **Step Timers**: Individual timeouts for each connection phase
+ *
+ * - All errors result in connection termination
+ *
+ * All operations are serialized using boost::asio::strand to ensure thread
+ * safety. The class is designed to be used exclusively within the ASIO event
+ * loop.
+ *
+ * @note This class should not be used directly. It is managed by OverlayImpl
+ *       as part of the peer discovery and connection management system.
+ *
+ */
 class ConnectAttempt : public OverlayImpl::Child,
                        public std::enable_shared_from_this<ConnectAttempt>
 {
 private:
     using error_code = boost::system::error_code;
-
     using endpoint_type = boost::asio::ip::tcp::endpoint;
-
     using request_type =
         boost::beast::http::request<boost::beast::http::empty_body>;
-
     using response_type =
         boost::beast::http::response<boost::beast::http::dynamic_body>;
-
     using socket_type = boost::asio::ip::tcp::socket;
     using middle_type = boost::beast::tcp_stream;
     using stream_type = boost::beast::ssl_stream<middle_type>;
     using shared_context = std::shared_ptr<boost::asio::ssl::context>;
 
+    /**
+     * @enum ConnectionStep
+     * @brief Represents the current phase of the connection establishment
+     * process
+     *
+     * Used for tracking progress and providing detailed timeout diagnostics.
+     * Each step has its own timeout value defined in StepTimeouts.
+     */
+    enum class ConnectionStep {
+        Init,            // Initial state, nothing started
+        TcpConnect,      // Establishing TCP connection to remote peer
+        TlsHandshake,    // Performing SSL/TLS handshake
+        HttpWrite,       // Sending HTTP upgrade request
+        HttpRead,        // Reading HTTP upgrade response
+        Complete,        // Connection successfully established
+        ShutdownStarted  // Connection shutdown has started
+    };
+
+    // A timeout for connection process, greater than all step timeouts
+    static constexpr std::chrono::seconds connectTimeout{25};
+
+    /**
+     * @struct StepTimeouts
+     * @brief Defines timeout values for each connection step
+     *
+     * These timeouts are designed to detect slow individual phases while
+     * allowing the global timeout to enforce the overall time limit.
+     */
+    struct StepTimeouts
+    {
+        // TCP connection timeout
+        static constexpr std::chrono::seconds tcpConnect{8};
+        // SSL handshake timeout
+        static constexpr std::chrono::seconds tlsHandshake{8};
+        // HTTP write timeout
+        static constexpr std::chrono::seconds httpWrite{3};
+        // HTTP read timeout
+        static constexpr std::chrono::seconds httpRead{3};
+        // SSL shutdown timeout
+        static constexpr std::chrono::seconds tlsShutdown{2};
+    };
+
+    // Core application and networking components
     Application& app_;
-    std::uint32_t const id_;
+    Peer::id_t const id_;
     beast::WrappedSink sink_;
     beast::Journal const journal_;
     endpoint_type remote_endpoint_;
     Resource::Consumer usage_;
+
     boost::asio::strand<boost::asio::io_context::executor_type> strand_;
     boost::asio::basic_waitable_timer<std::chrono::steady_clock> timer_;
-    std::unique_ptr<stream_type> stream_ptr_;
+    boost::asio::basic_waitable_timer<std::chrono::steady_clock> stepTimer_;
+
+    std::unique_ptr<stream_type> stream_ptr_;  // SSL stream (owned)
     socket_type& socket_;
     stream_type& stream_;
     boost::beast::multi_buffer read_buf_;
+
     response_type response_;
     std::shared_ptr<PeerFinder::Slot> slot_;
     request_type req_;
 
+    bool shutdown_ = false;   // Shutdown has been initiated
+    bool ioPending_ = false;  // Async I/O operation in progress
+    ConnectionStep currentStep_ = ConnectionStep::Init;
+
 public:
+    /**
+     * @brief Construct a new ConnectAttempt object
+     *
+     * @param app Application context providing configuration and services
+     * @param io_context ASIO I/O context for async operations
+     * @param remote_endpoint Target peer endpoint to connect to
+     * @param usage Resource usage tracker for rate limiting
+     * @param context Shared SSL context for encryption
+     * @param id Unique peer identifier for this connection attempt
+     * @param slot PeerFinder slot representing this connection
+     * @param journal Logging interface for diagnostics
+     * @param overlay Parent overlay manager
+     *
+     * @note The constructor only initializes the object. Call run() to begin
+     *       the actual connection attempt.
+     */
     ConnectAttempt(
         Application& app,
         boost::asio::io_context& io_context,
         endpoint_type const& remote_endpoint,
         Resource::Consumer usage,
         shared_context const& context,
-        std::uint32_t id,
+        Peer::id_t id,
         std::shared_ptr<PeerFinder::Slot> const& slot,
         beast::Journal journal,
         OverlayImpl& overlay);
 
     ~ConnectAttempt();
 
+    /**
+     * @brief Stop the connection attempt
+     *
+     * This method is thread-safe and can be called from any thread.
+     */
     void
     stop() override;
 
+    /**
+     * @brief Begin the connection attempt
+     *
+     * This method is thread-safe and posts to the strand if needed.
+     */
     void
     run();
 
 private:
+    /**
+     * @brief Set timers for the specified connection step
+     *
+     * @param step The connection step to set timers for
+     *
+     * Sets both the step-specific timer and the global timer (if not already
+     * set).
+     */
     void
-    close();
-    void
-    fail(std::string const& reason);
-    void
-    fail(std::string const& name, error_code ec);
-    void
-    setTimer();
+    setTimer(ConnectionStep step);
+
+    /**
+     * @brief Cancel both global and step timers
+     *
+     * Used during cleanup and when connection completes successfully.
+     * Exceptions from timer cancellation are safely ignored.
+     */
     void
     cancelTimer();
+
+    /**
+     * @brief Handle timer expiration events
+     *
+     * @param ec Error code from timer operation
+     *
+     * Determines which timer expired (global vs step) and logs appropriate
+     * diagnostic information before terminating the connection.
+     */
     void
     onTimer(error_code ec);
+
+    // Connection phase handlers
     void
-    onConnect(error_code ec);
+    onConnect(error_code ec);  // TCP connection completion handler
     void
-    onHandshake(error_code ec);
+    onHandshake(error_code ec);  // TLS handshake completion handler
     void
-    onWrite(error_code ec);
+    onWrite(error_code ec);  // HTTP write completion handler
     void
-    onRead(error_code ec);
+    onRead(error_code ec);  // HTTP read completion handler
+
+    // Error and cleanup handlers
     void
-    onShutdown(error_code ec);
+    fail(std::string const& reason);  // Fail with custom reason
+    void
+    fail(std::string const& name, error_code ec);  // Fail with system error
+    void
+    shutdown();  // Initiate graceful shutdown
+    void
+    tryAsyncShutdown();  // Attempt async SSL shutdown
+    void
+    onShutdown(error_code ec);  // SSL shutdown completion handler
+    void
+    close();  // Force close socket
+
+    /**
+     * @brief Process the HTTP upgrade response from peer
+     *
+     * Validates the peer's response, extracts protocol information,
+     * verifies handshake, and either creates a PeerImp or handles
+     * redirect responses.
+     */
     void
     processResponse();
 
+    static std::string
+    stepToString(ConnectionStep step)
+    {
+        switch (step)
+        {
+            case ConnectionStep::Init:
+                return "Init";
+            case ConnectionStep::TcpConnect:
+                return "TcpConnect";
+            case ConnectionStep::TlsHandshake:
+                return "TlsHandshake";
+            case ConnectionStep::HttpWrite:
+                return "HttpWrite";
+            case ConnectionStep::HttpRead:
+                return "HttpRead";
+            case ConnectionStep::Complete:
+                return "Complete";
+            case ConnectionStep::ShutdownStarted:
+                return "ShutdownStarted";
+        }
+        return "Unknown";
+    };
+
     template <class = void>
     static boost::asio::ip::tcp::endpoint
     parse_endpoint(std::string const& s, boost::system::error_code& ec)

src/xrpld/overlay/detail/PeerImp.cpp

@@ -44,6 +44,7 @@
 #include <boost/beast/core/ostream.hpp>
 
 #include <algorithm>
+#include <chrono>
 #include <memory>
 #include <mutex>
 #include <numeric>
@@ -59,6 +60,10 @@ std::chrono::milliseconds constexpr peerHighLatency{300};
 
 /** How often we PING the peer to check for latency and sendq probe */
 std::chrono::seconds constexpr peerTimerInterval{60};
+
+/** The timeout for a shutdown timer */
+std::chrono::seconds constexpr shutdownTimerInterval{5};
+
 }  // namespace
 
 // TODO: Remove this exclusion once unit tests are added after the hotfix
@@ -215,23 +220,17 @@ PeerImp::stop()
 {
     if (!strand_.running_in_this_thread())
         return post(strand_, std::bind(&PeerImp::stop, shared_from_this()));
-    if (socket_.is_open())
-    {
-        // The rationale for using different severity levels is that
-        // outbound connections are under our control and may be logged
-        // at a higher level, but inbound connections are more numerous and
-        // uncontrolled so to prevent log flooding the severity is reduced.
-        //
-        if (inbound_)
-        {
-            JLOG(journal_.debug()) << "Stop";
-        }
-        else
-        {
-            JLOG(journal_.info()) << "Stop";
-        }
-    }
-    close();
+
+    if (!socket_.is_open())
+        return;
+
+    // The rationale for using different severity levels is that
+    // outbound connections are under our control and may be logged
+    // at a higher level, but inbound connections are more numerous and
+    // uncontrolled so to prevent log flooding the severity is reduced.
+    JLOG(journal_.debug()) << "stop: Stop";
+
+    shutdown();
 }
 
 //------------------------------------------------------------------------------
@@ -241,11 +240,14 @@ PeerImp::send(std::shared_ptr<Message> const& m)
 {
     if (!strand_.running_in_this_thread())
         return post(strand_, std::bind(&PeerImp::send, shared_from_this(), m));
-    if (gracefulClose_)
-        return;
-    if (detaching_)
+
+    if (!socket_.is_open())
         return;
 
+    // we are in progress of closing the connection
+    if (shutdown_)
+        return tryAsyncShutdown();
+
     auto validator = m->getValidatorKey();
     if (validator && !squelch_.expireSquelch(*validator))
     {
@@ -287,6 +289,7 @@ PeerImp::send(std::shared_ptr<Message> const& m)
     if (sendq_size != 0)
         return;
 
+    writePending_ = true;
     boost::asio::async_write(
         stream_,
         boost::asio::buffer(
@@ -573,34 +576,21 @@ PeerImp::hasRange(std::uint32_t uMin, std::uint32_t uMax)
 //------------------------------------------------------------------------------
 
 void
-PeerImp::close()
+PeerImp::fail(std::string const& name, error_code ec)
 {
     XRPL_ASSERT(
         strand_.running_in_this_thread(),
-        "ripple::PeerImp::close : strand in this thread");
-    if (socket_.is_open())
-    {
-        detaching_ = true;  // DEPRECATED
-        try
-        {
-            timer_.cancel();
-            socket_.close();
-        }
-        catch (boost::system::system_error const&)
-        {
-            // ignored
-        }
+        "ripple::PeerImp::fail : strand in this thread");
 
-        overlay_.incPeerDisconnect();
-        if (inbound_)
-        {
-            JLOG(journal_.debug()) << "Closed";
-        }
-        else
-        {
-            JLOG(journal_.info()) << "Closed";
-        }
-    }
+    if (!socket_.is_open())
+        return;
+
+    JLOG(journal_.warn()) << name << " from "
+                          << toBase58(TokenType::NodePublic, publicKey_)
+                          << " at " << remote_address_.to_string() << ": "
+                          << ec.message();
+
+    shutdown();
 }
 
 void
@@ -613,45 +603,39 @@ PeerImp::fail(std::string const& reason)
                 (void(Peer::*)(std::string const&)) & PeerImp::fail,
                 shared_from_this(),
                 reason));
-    if (journal_.active(beast::severities::kWarning) && socket_.is_open())
+
+    if (!socket_.is_open())
+        return;
+
+    // Call to name() locks, log only if the message will be outputed
+    if (journal_.active(beast::severities::kWarning))
     {
         std::string const n = name();
         JLOG(journal_.warn()) << (n.empty() ? remote_address_.to_string() : n)
                               << " failed: " << reason;
     }
-    close();
+
+    shutdown();
 }
 
 void
-PeerImp::fail(std::string const& name, error_code ec)
+PeerImp::tryAsyncShutdown()
 {
     XRPL_ASSERT(
         strand_.running_in_this_thread(),
-        "ripple::PeerImp::fail : strand in this thread");
-    if (socket_.is_open())
-    {
-        JLOG(journal_.warn())
-            << name << " from " << toBase58(TokenType::NodePublic, publicKey_)
-            << " at " << remote_address_.to_string() << ": " << ec.message();
-    }
-    close();
-}
+        "ripple::PeerImp::tryAsyncShutdown : strand in this thread");
 
-void
-PeerImp::gracefulClose()
-{
-    XRPL_ASSERT(
-        strand_.running_in_this_thread(),
-        "ripple::PeerImp::gracefulClose : strand in this thread");
-    XRPL_ASSERT(
-        socket_.is_open(), "ripple::PeerImp::gracefulClose : socket is open");
-    XRPL_ASSERT(
-        !gracefulClose_,
-        "ripple::PeerImp::gracefulClose : socket is not closing");
-    gracefulClose_ = true;
-    if (send_queue_.size() > 0)
+    if (!shutdown_ || shutdownStarted_)
         return;
-    setTimer();
+
+    if (readPending_ || writePending_)
+        return;
+
+    shutdownStarted_ = true;
+
+    setTimer(shutdownTimerInterval);
+
+    // gracefully shutdown the SSL socket, performing a shutdown handshake
     stream_.async_shutdown(bind_executor(
         strand_,
         std::bind(
@@ -659,69 +643,125 @@ PeerImp::gracefulClose()
 }
 
 void
-PeerImp::setTimer()
+PeerImp::shutdown()
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::shutdown: strand in this thread");
+
+    if (!socket_.is_open() || shutdown_)
+        return;
+
+    shutdown_ = true;
+
+    boost::beast::get_lowest_layer(stream_).cancel();
+
+    tryAsyncShutdown();
+}
+
+void
+PeerImp::onShutdown(error_code ec)
+{
+    cancelTimer();
+    if (ec)
+    {
+        // - eof: the stream was cleanly closed
+        // - operation_aborted: an expired timer (slow shutdown)
+        // - stream_truncated: the tcp connection closed (no handshake) it could
+        // occur if a peer does not perform a graceful disconnect
+        // - broken_pipe: the peer is gone
+        bool shouldLog =
+            (ec != boost::asio::error::eof &&
+             ec != boost::asio::error::operation_aborted &&
+             ec.message().find("application data after close notify") ==
+                 std::string::npos);
+
+        if (shouldLog)
+        {
+            JLOG(journal_.debug()) << "onShutdown: " << ec.message();
+        }
+    }
+
+    close();
+}
+
+void
+PeerImp::close()
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::close : strand in this thread");
+
+    if (!socket_.is_open())
+        return;
+
+    cancelTimer();
+
+    error_code ec;
+    socket_.close(ec);
+
+    overlay_.incPeerDisconnect();
+
+    // The rationale for using different severity levels is that
+    // outbound connections are under our control and may be logged
+    // at a higher level, but inbound connections are more numerous and
+    // uncontrolled so to prevent log flooding the severity is reduced.
+    JLOG((inbound_ ? journal_.debug() : journal_.info())) << "close: Closed";
+}
+
+//------------------------------------------------------------------------------
+
+void
+PeerImp::setTimer(std::chrono::seconds interval)
 {
     try
     {
-        timer_.expires_after(peerTimerInterval);
+        timer_.expires_after(interval);
     }
-    catch (boost::system::system_error const& e)
+    catch (std::exception const& ex)
     {
-        JLOG(journal_.error()) << "setTimer: " << e.code();
-        return;
+        JLOG(journal_.error()) << "setTimer: " << ex.what();
+        return shutdown();
     }
+
     timer_.async_wait(bind_executor(
         strand_,
         std::bind(
             &PeerImp::onTimer, shared_from_this(), std::placeholders::_1)));
 }
 
-// convenience for ignoring the error code
-void
-PeerImp::cancelTimer()
-{
-    try
-    {
-        timer_.cancel();
-    }
-    catch (boost::system::system_error const&)
-    {
-        // ignored
-    }
-}
-
-//------------------------------------------------------------------------------
-
-std::string
-PeerImp::makePrefix(id_t id)
-{
-    std::stringstream ss;
-    ss << "[" << std::setfill('0') << std::setw(3) << id << "] ";
-    return ss.str();
-}
-
 void
 PeerImp::onTimer(error_code const& ec)
 {
-    if (!socket_.is_open())
-        return;
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::onTimer : strand in this thread");
 
-    if (ec == boost::asio::error::operation_aborted)
+    if (!socket_.is_open())
         return;
 
     if (ec)
     {
+        // do not initiate shutdown, timers are frequently cancelled
+        if (ec == boost::asio::error::operation_aborted)
+            return;
+
         // This should never happen
         JLOG(journal_.error()) << "onTimer: " << ec.message();
         return close();
     }
 
-    if (large_sendq_++ >= Tuning::sendqIntervals)
+    // the timer expired before the shutdown completed
+    // force close the connection
+    if (shutdown_)
     {
-        fail("Large send queue");
-        return;
+        JLOG(journal_.debug()) << "onTimer: shutdown timer expired";
+        return close();
     }
 
+    if (large_sendq_++ >= Tuning::sendqIntervals)
+        return fail("Large send queue");
+
     if (auto const t = tracking_.load(); !inbound_ && t != Tracking::converged)
     {
         clock_type::duration duration;
@@ -737,17 +777,13 @@ PeerImp::onTimer(error_code const& ec)
              (duration > app_.config().MAX_UNKNOWN_TIME)))
         {
             overlay_.peerFinder().on_failure(slot_);
-            fail("Not useful");
-            return;
+            return fail("Not useful");
         }
     }
 
     // Already waiting for PONG
     if (lastPingSeq_)
-    {
-        fail("Ping Timeout");
-        return;
-    }
+        return fail("Ping Timeout");
 
     lastPingTime_ = clock_type::now();
     lastPingSeq_ = rand_int<std::uint32_t>();
@@ -758,22 +794,28 @@ PeerImp::onTimer(error_code const& ec)
 
     send(std::make_shared<Message>(message, protocol::mtPING));
 
-    setTimer();
+    setTimer(peerTimerInterval);
 }
 
 void
-PeerImp::onShutdown(error_code ec)
+PeerImp::cancelTimer() noexcept
 {
-    cancelTimer();
-    // If we don't get eof then something went wrong
-    if (!ec)
+    try
     {
-        JLOG(journal_.error()) << "onShutdown: expected error condition";
-        return close();
+        timer_.cancel();
     }
-    if (ec != boost::asio::error::eof)
-        return fail("onShutdown", ec);
-    close();
+    catch (std::exception const& ex)
+    {
+        JLOG(journal_.error()) << "cancelTimer: " << ex.what();
+    }
+}
+
+std::string
+PeerImp::makePrefix(id_t id)
+{
+    std::stringstream ss;
+    ss << "[" << std::setfill('0') << std::setw(3) << id << "] ";
+    return ss.str();
 }
 
 //------------------------------------------------------------------------------
@@ -786,6 +828,10 @@ PeerImp::doAccept()
 
     JLOG(journal_.debug()) << "doAccept: " << remote_address_;
 
+    // a shutdown was initiated before the handshake, there is nothing to do
+    if (shutdown_)
+        return tryAsyncShutdown();
+
     auto const sharedValue = makeSharedValue(*stream_ptr_, journal_);
 
     // This shouldn't fail since we already computed
@@ -793,7 +839,7 @@ PeerImp::doAccept()
     if (!sharedValue)
         return fail("makeSharedValue: Unexpected failure");
 
-    JLOG(journal_.info()) << "Protocol: " << to_string(protocol_);
+    JLOG(journal_.debug()) << "Protocol: " << to_string(protocol_);
     JLOG(journal_.info()) << "Public Key: "
                           << toBase58(TokenType::NodePublic, publicKey_);
 
@@ -836,7 +882,7 @@ PeerImp::doAccept()
                 if (!socket_.is_open())
                     return;
                 if (ec == boost::asio::error::operation_aborted)
-                    return;
+                    return tryAsyncShutdown();
                 if (ec)
                     return fail("onWriteResponse", ec);
                 if (write_buffer->size() == bytes_transferred)
@@ -865,6 +911,10 @@ PeerImp::domain() const
 void
 PeerImp::doProtocolStart()
 {
+    // a shutdown was initiated before the handshare, there is nothing to do
+    if (shutdown_)
+        return tryAsyncShutdown();
+
     onReadMessage(error_code(), 0);
 
     // Send all the validator lists that have been loaded
@@ -896,30 +946,45 @@ PeerImp::doProtocolStart()
     if (auto m = overlay_.getManifestsMessage())
         send(m);
 
-    setTimer();
+    setTimer(peerTimerInterval);
 }
 
 // Called repeatedly with protocol message data
 void
 PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
 {
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::onReadMessage : strand in this thread");
+
+    readPending_ = false;
+
     if (!socket_.is_open())
         return;
-    if (ec == boost::asio::error::operation_aborted)
-        return;
-    if (ec == boost::asio::error::eof)
-    {
-        JLOG(journal_.info()) << "EOF";
-        return gracefulClose();
-    }
+
     if (ec)
+    {
+        if (ec == boost::asio::error::eof)
+        {
+            JLOG(journal_.debug()) << "EOF";
+            return shutdown();
+        }
+
+        if (ec == boost::asio::error::operation_aborted)
+            return tryAsyncShutdown();
+
         return fail("onReadMessage", ec);
+    }
+    // we started shutdown, no reason to process further data
+    if (shutdown_)
+        return tryAsyncShutdown();
+
     if (auto stream = journal_.trace())
     {
-        if (bytes_transferred > 0)
-            stream << "onReadMessage: " << bytes_transferred << " bytes";
-        else
-            stream << "onReadMessage";
+        stream << "onReadMessage: "
+               << (bytes_transferred > 0
+                       ? to_string(bytes_transferred) + " bytes"
+                       : "");
     }
 
     metrics_.recv.add_message(bytes_transferred);
@@ -941,17 +1006,29 @@ PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
             350ms,
             journal_);
 
-        if (ec)
-            return fail("onReadMessage", ec);
         if (!socket_.is_open())
             return;
-        if (gracefulClose_)
-            return;
+
+        // the error_code is produced by invokeProtocolMessage
+        // it could be due to a bad message
+        if (ec)
+            return fail("onReadMessage", ec);
+
         if (bytes_consumed == 0)
             break;
+
         read_buffer_.consume(bytes_consumed);
     }
 
+    // check if a shutdown was initiated while processing messages
+    if (shutdown_)
+        return tryAsyncShutdown();
+
+    readPending_ = true;
+
+    XRPL_ASSERT(
+        !shutdownStarted_, "ripple::PeerImp::onReadMessage : shutdown started");
+
     // Timeout on writes only
     stream_.async_read_some(
         read_buffer_.prepare(std::max(Tuning::readBufferBytes, hint)),
@@ -967,18 +1044,29 @@ PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
 void
 PeerImp::onWriteMessage(error_code ec, std::size_t bytes_transferred)
 {
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::onWriteMessage : strand in this thread");
+
+    writePending_ = false;
+
     if (!socket_.is_open())
         return;
-    if (ec == boost::asio::error::operation_aborted)
-        return;
+
     if (ec)
+    {
+        if (ec == boost::asio::error::operation_aborted)
+            return tryAsyncShutdown();
+
         return fail("onWriteMessage", ec);
+    }
+
     if (auto stream = journal_.trace())
     {
-        if (bytes_transferred > 0)
-            stream << "onWriteMessage: " << bytes_transferred << " bytes";
-        else
-            stream << "onWriteMessage";
+        stream << "onWriteMessage: "
+               << (bytes_transferred > 0
+                       ? to_string(bytes_transferred) + " bytes"
+                       : "");
     }
 
     metrics_.sent.add_message(bytes_transferred);
@@ -987,8 +1075,17 @@ PeerImp::onWriteMessage(error_code ec, std::size_t bytes_transferred)
         !send_queue_.empty(),
         "ripple::PeerImp::onWriteMessage : non-empty send buffer");
     send_queue_.pop();
+
+    if (shutdown_)
+        return tryAsyncShutdown();
+
     if (!send_queue_.empty())
     {
+        writePending_ = true;
+        XRPL_ASSERT(
+            !shutdownStarted_,
+            "ripple::PeerImp::onWriteMessage : shutdown started");
+
         // Timeout on writes only
         return boost::asio::async_write(
             stream_,
@@ -1002,16 +1099,6 @@ PeerImp::onWriteMessage(error_code ec, std::size_t bytes_transferred)
                     std::placeholders::_1,
                     std::placeholders::_2)));
     }
-
-    if (gracefulClose_)
-    {
-        return stream_.async_shutdown(bind_executor(
-            strand_,
-            std::bind(
-                &PeerImp::onShutdown,
-                shared_from_this(),
-                std::placeholders::_1)));
-    }
 }
 
 //------------------------------------------------------------------------------
@@ -2880,6 +2967,9 @@ PeerImp::checkTransaction(
             (stx->getFieldU32(sfLastLedgerSequence) <
              app_.getLedgerMaster().getValidLedgerIndex()))
         {
+            JLOG(p_journal_.info())
+                << "Marking transaction " << stx->getTransactionID()
+                << "as BAD because it's expired";
             app_.getHashRouter().setFlags(
                 stx->getTransactionID(), HashRouterFlags::BAD);
             charge(Resource::feeUselessData, "expired tx");
@@ -2936,7 +3026,7 @@ PeerImp::checkTransaction(
             {
                 if (!validReason.empty())
                 {
-                    JLOG(p_journal_.trace())
+                    JLOG(p_journal_.debug())
                         << "Exception checking transaction: " << validReason;
                 }
 
@@ -2963,7 +3053,7 @@ PeerImp::checkTransaction(
         {
             if (!reason.empty())
             {
-                JLOG(p_journal_.trace())
+                JLOG(p_journal_.debug())
                     << "Exception checking transaction: " << reason;
             }
             app_.getHashRouter().setFlags(

src/xrpld/overlay/detail/PeerImp.h

@@ -40,6 +40,7 @@
 #include <boost/endian/conversion.hpp>
 #include <boost/thread/shared_mutex.hpp>
 
+#include <atomic>
 #include <cstdint>
 #include <optional>
 #include <queue>
@@ -49,6 +50,68 @@ namespace ripple {
 struct ValidatorBlobInfo;
 class SHAMap;
 
+/**
+ * @class PeerImp
+ * @brief This class manages established peer-to-peer connections, handles
+ message exchange, monitors connection health, and graceful shutdown.
+ *
+
+ * The PeerImp shutdown mechanism is a multi-stage process
+ * designed to ensure graceful connection termination while handling ongoing
+ * I/O operations safely. The shutdown can be initiated from multiple points
+ * and follows a deterministic state machine.
+ *
+ * The shutdown process can be triggered from several entry points:
+ * - **External requests**: `stop()` method called by overlay management
+ * - **Error conditions**: `fail(error_code)` or `fail(string)` on protocol
+ * violations
+ * - **Timer expiration**: Various timeout scenarios (ping timeout, large send
+ * queue)
+ * - **Connection health**: Peer tracking divergence or unknown state timeouts
+ *
+ * The shutdown follows this progression:
+ *
+ * Normal Operation → shutdown() → tryAsyncShutdown() → onShutdown() → close()
+ *                      ↓              ↓                 ↓              ↓
+ *                 Set shutdown_   SSL graceful      Timer cancel   Socket close
+ *                 Cancel timer    shutdown start    & cleanup      & metrics
+ *                 5s safety timer Set shutdownStarted_              update
+ *
+ * Two primary flags coordinate the shutdown process:
+ * - `shutdown_`: Set when shutdown is requested
+ * - `shutdownStarted_`: Set when SSL shutdown begins
+ *
+ * The shutdown mechanism carefully coordinates with ongoing read/write
+ * operations:
+ *
+ * **Read Operations (`onReadMessage`)**:
+ * - Checks `shutdown_` flag after processing each message batch
+ * - If shutdown initiated during processing, calls `tryAsyncShutdown()`
+ *
+ * **Write Operations (`onWriteMessage`)**:
+ * - Checks `shutdown_` flag before queuing new writes
+ * - Calls `tryAsyncShutdown()` when shutdown flag detected
+ *
+ * Multiple timers require coordination during shutdown:
+ * 1. **Peer Timer**: Regular ping/pong timer cancelled immediately in
+ * `shutdown()`
+ * 2. **Shutdown Timer**: 5-second safety timer ensures shutdown completion
+ * 3. **Operation Cancellation**: All pending async operations are cancelled
+ *
+ * The shutdown implements fallback mechanisms:
+ * - **Graceful Path**: SSL shutdown → Socket close → Cleanup
+ * - **Forced Path**: If SSL shutdown fails or times out, proceeds to socket
+ * close
+ * - **Safety Timer**: 5-second timeout prevents hanging shutdowns
+ *
+ * All shutdown operations are serialized through the boost::asio::strand to
+ * ensure thread safety. The strand guarantees that shutdown state changes
+ * and I/O operation callbacks are executed sequentially.
+ *
+ * @note This class requires careful coordination between async operations,
+ * timer management, and shutdown procedures to ensure no resource leaks
+ * or hanging connections in high-throughput networking scenarios.
+ */
 class PeerImp : public Peer,
                 public std::enable_shared_from_this<PeerImp>,
                 public OverlayImpl::Child
@@ -79,6 +142,8 @@ private:
     socket_type& socket_;
     stream_type& stream_;
     boost::asio::strand<boost::asio::executor> strand_;
+
+    // Multi-purpose timer for peer activity monitoring and shutdown safety
     waitable_timer timer_;
 
     // Updated at each stage of the connection process to reflect
@@ -95,7 +160,6 @@ private:
 
     std::atomic<Tracking> tracking_;
     clock_type::time_point trackingTime_;
-    bool detaching_ = false;
     // Node public key of peer.
     PublicKey const publicKey_;
     std::string name_;
@@ -175,7 +239,19 @@ private:
     http_response_type response_;
     boost::beast::http::fields const& headers_;
     std::queue<std::shared_ptr<Message>> send_queue_;
-    bool gracefulClose_ = false;
+
+    // Primary shutdown flag set when shutdown is requested
+    bool shutdown_ = false;
+
+    // SSL shutdown coordination flag
+    bool shutdownStarted_ = false;
+
+    // Indicates a read operation is currently pending
+    bool readPending_ = false;
+
+    // Indicates a write operation is currently pending
+    bool writePending_ = false;
+
     int large_sendq_ = 0;
     std::unique_ptr<LoadEvent> load_event_;
     // The highest sequence of each PublisherList that has
@@ -425,9 +501,6 @@ public:
     bool
     isHighLatency() const override;
 
-    void
-    fail(std::string const& reason);
-
     bool
     compressionEnabled() const override
     {
@@ -441,32 +514,129 @@ public:
     }
 
 private:
-    void
-    close();
-
+    /**
+     * @brief Handles a failure associated with a specific error code.
+     *
+     * This function is called when an operation fails with an error code. It
+     * logs the warning message and gracefully shutdowns the connection.
+     *
+     * The function will do nothing if the connection is already closed or if a
+     * shutdown is already in progress.
+     *
+     * @param name The name of the operation that failed (e.g., "read",
+     * "write").
+     * @param ec The error code associated with the failure.
+     * @note This function must be called from within the object's strand.
+     */
     void
     fail(std::string const& name, error_code ec);
 
+    /**
+     * @brief Handles a failure described by a reason string.
+     *
+     * This overload is used for logical errors or protocol violations not
+     * associated with a specific error code. It logs a warning with the
+     * given reason, then initiates a graceful shutdown.
+     *
+     * The function will do nothing if the connection is already closed or if a
+     * shutdown is already in progress.
+     *
+     * @param reason A descriptive string explaining the reason for the failure.
+     * @note This function must be called from within the object's strand.
+     */
     void
-    gracefulClose();
+    fail(std::string const& reason);
 
+    /** @brief Initiates the peer disconnection sequence.
+     *
+     * This is the primary entry point to start closing a peer connection. It
+     * marks the peer for shutdown and cancels any outstanding asynchronous
+     * operations. This cancellation allows the graceful shutdown to proceed
+     * once the handlers for the cancelled operations have completed.
+     *
+     * @note This method must be called on the peer's strand.
+     */
     void
-    setTimer();
+    shutdown();
 
+    /** @brief Attempts to perform a graceful SSL shutdown if conditions are
+     * met.
+     *
+     * This helper function checks if the peer is in a state where a graceful
+     * SSL shutdown can be performed (i.e., shutdown has been requested and no
+     * I/O operations are currently in progress).
+     *
+     * @note This method must be called on the peer's strand.
+     */
     void
-    cancelTimer();
+    tryAsyncShutdown();
+
+    /**
+     * @brief Handles the completion of the asynchronous SSL shutdown.
+     *
+     * This function is the callback for the `async_shutdown` operation started
+     * in `shutdown()`. Its first action is to cancel the timer. It
+     * then inspects the error code to determine the outcome.
+     *
+     * Regardless of the result, this function proceeds to call `close()` to
+     * ensure the underlying socket is fully closed.
+     *
+     * @param ec The error code resulting from the `async_shutdown` operation.
+     */
+    void
+    onShutdown(error_code ec);
+
+    /**
+     * @brief Forcibly closes the underlying socket connection.
+     *
+     * This function provides the final, non-graceful shutdown of the peer
+     * connection. It ensures any pending timers are cancelled and then
+     * immediately closes the TCP socket, bypassing the SSL shutdown handshake.
+     *
+     * After closing, it notifies the overlay manager of the disconnection.
+     *
+     * @note This function must be called from within the object's strand.
+     */
+    void
+    close();
+
+    /**
+     * @brief Sets and starts the peer timer.
+     *
+     * This function starts timer, which is used to detect inactivity
+     * and prevent stalled connections. It sets the timer to expire after the
+     * predefined `peerTimerInterval`.
+     *
+     * @note This function will terminate the connection in case of any errors.
+     */
+    void
+    setTimer(std::chrono::seconds interval);
+
+    /**
+     * @brief Handles the expiration of the peer activity timer.
+     *
+     * This callback is invoked when the timer set by `setTimer` expires. It
+     * watches the peer connection, checking for various timeout and health
+     * conditions.
+     *
+     * @param ec The error code associated with the timer's expiration.
+     * `operation_aborted` is expected if the timer was cancelled.
+     */
+    void
+    onTimer(error_code const& ec);
+
+    /**
+     * @brief Cancels any pending wait on the peer activity timer.
+     *
+     * This function is called to stop the timer. It gracefully manages any
+     * errors that might occur during the cancellation process.
+     */
+    void
+    cancelTimer() noexcept;
 
     static std::string
     makePrefix(id_t id);
 
-    // Called when the timer wait completes
-    void
-    onTimer(boost::system::error_code const& ec);
-
-    // Called when SSL shutdown completes
-    void
-    onShutdown(error_code ec);
-
     void
     doAccept();