ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-11-04 11:15:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into ximinez/lending-refactoring-1

Browse Source
This commit is contained in:
Ed Hennis
2025-09-16 10:46:41 -04:00
committed by GitHub
parent 91031dbec4 17a2606591
commit 935749fe1f
18 changed files with 2186 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/on-pr.yml vendored
View File

@@ -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 }}

4
.github/workflows/upload-conan-deps.yml vendored
View File

@@ -83,9 +83,9 @@ jobs:
force_build: ${{ github.event_name == 'schedule' || github.event.inputs.force_source_build == 'true' }}
- name: Log into Conan remote
if: github.repository_owner == 'XRPLF' && github.event_name != 'pull_request'
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'
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' || '' }}

9
include/xrpl/protocol/LedgerFormats.h
View File

@@ -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,

33
include/xrpl/protocol/TxFlags.h
View File

@@ -154,6 +154,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);
@@ -164,6 +178,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;

1
include/xrpl/protocol/detail/features.macro
View File

@@ -32,6 +32,7 @@
// 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)

1
include/xrpl/protocol/detail/ledger_entries.macro
View File

@@ -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

1
include/xrpl/protocol/detail/sfields.macro
View File

@@ -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)

4
include/xrpl/protocol/detail/transactions.macro
View File

@@ -711,6 +711,7 @@ TRANSACTION(ttMPTOKEN_ISSUANCE_CREATE, 54, MPTokenIssuanceCreate,
{sfMaximumAmount, soeOPTIONAL},
{sfMPTokenMetadata, soeOPTIONAL},
{sfDomainID, soeOPTIONAL},
{sfMutableFlags, soeOPTIONAL},
}))
/** This transaction type destroys a MPTokensIssuance instance */
@@ -737,6 +738,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 */

920
src/test/app/MPToken_test.cpp
View File

@@ -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);
}
};

93
src/test/jtx/impl/mpt.cpp
View File

@@ -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,

16
src/test/jtx/mpt.h
View File

@@ -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
{

12
src/xrpld/app/tx/detail/MPTokenIssuanceCreate.cpp
View File

@@ -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();
}

1
src/xrpld/app/tx/detail/MPTokenIssuanceCreate.h
View File

@@ -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

150
src/xrpld/app/tx/detail/MPTokenIssuanceSet.cpp
View File

@@ -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.

488
src/xrpld/overlay/detail/ConnectAttempt.cpp
View File

@@ -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_,

208
src/xrpld/overlay/detail/ConnectAttempt.h
View File

@@ -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)

397
src/xrpld/overlay/detail/PeerImp.cpp
View File

@@ -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)));
}
}
//------------------------------------------------------------------------------

208
src/xrpld/overlay/detail/PeerImp.h
View File

@@ -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();