feat: Add DatagramMonitor, a UDP node-stats exporter

.github/workflows/reusable-package.yml

@@ -25,7 +25,7 @@ on:
         description: "The base URL of the Nexus instance hosting the deb and rpm repositories."
         required: false
         type: string
-        default: https://packages.xrplf.org
+        default: https://packages-upload.xrplf.org
 
     secrets:
       remote_username:
@@ -112,7 +112,7 @@ jobs:
         env:
           PACKAGE_TYPE: ${{ matrix.package_type }}
           PACKAGE_VARIANT: ${{ matrix.package_variant }}
-          PKG_RELEASE: ${{ steps.release_info.outputs.pkg_release }}
+          PKG_RELEASE: ${{ startsWith(github.ref, 'refs/tags/') && '2' || steps.release_info.outputs.pkg_release }}
           CHANNEL: ${{ steps.release_info.outputs.channel }}
         run: |
           ./package/build_pkg.py \

include/xrpl/basics/MathUtilities.h

@@ -3,9 +3,84 @@
 #include <algorithm>
 #include <cassert>
 #include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <optional>
 
 namespace xrpl {
 
+/**
+ * Add two signed 64-bit integers, returning std::nullopt when the exact
+ * mathematical sum is not representable in std::int64_t.
+ */
+[[nodiscard]] constexpr std::optional<std::int64_t>
+checkedAdd(std::int64_t a, std::int64_t b) noexcept
+{
+    using L = std::numeric_limits<std::int64_t>;
+
+    if ((b > 0 && a > L::max() - b) || (b < 0 && a < L::min() - b))
+        return std::nullopt;
+
+    return a + b;
+}
+
+/**
+ * Subtract two signed 64-bit integers, returning std::nullopt when the exact
+ * mathematical difference is not representable in std::int64_t.
+ */
+[[nodiscard]] constexpr std::optional<std::int64_t>
+checkedSub(std::int64_t a, std::int64_t b) noexcept
+{
+    using L = std::numeric_limits<std::int64_t>;
+
+    if ((b > 0 && a < L::min() + b) || (b < 0 && a > L::max() + b))
+        return std::nullopt;
+
+    return a - b;
+}
+
+static_assert(checkedAdd(0, 0) == 0);
+static_assert(checkedAdd(1, -1) == 0);
+static_assert(checkedAdd(-5, 2) == -3);
+static_assert(!checkedAdd(std::numeric_limits<std::int64_t>::max(), 1).has_value());
+static_assert(!checkedAdd(std::numeric_limits<std::int64_t>::min(), -1).has_value());
+static_assert(
+    checkedAdd(std::numeric_limits<std::int64_t>::max() - 1, 1) ==
+    std::numeric_limits<std::int64_t>::max());
+static_assert(
+    checkedAdd(
+        std::numeric_limits<std::int64_t>::min(),
+        std::numeric_limits<std::int64_t>::max()) == -1);
+static_assert(
+    checkedAdd(
+        std::numeric_limits<std::int64_t>::max(),
+        std::numeric_limits<std::int64_t>::min()) == -1);
+static_assert(
+    !checkedAdd(std::numeric_limits<std::int64_t>::max(), std::numeric_limits<std::int64_t>::max())
+         .has_value());
+static_assert(
+    !checkedAdd(std::numeric_limits<std::int64_t>::min(), std::numeric_limits<std::int64_t>::min())
+         .has_value());
+
+static_assert(checkedSub(0, 0) == 0);
+static_assert(checkedSub(1, 1) == 0);
+static_assert(checkedSub(-5, 2) == -7);
+static_assert(checkedSub(-5, -2) == -3);
+static_assert(!checkedSub(std::numeric_limits<std::int64_t>::min(), 1).has_value());
+static_assert(!checkedSub(std::numeric_limits<std::int64_t>::max(), -1).has_value());
+static_assert(
+    checkedSub(std::numeric_limits<std::int64_t>::min() + 1, 1) ==
+    std::numeric_limits<std::int64_t>::min());
+static_assert(
+    checkedSub(-1, std::numeric_limits<std::int64_t>::max()) ==
+    std::numeric_limits<std::int64_t>::min());
+static_assert(
+    !checkedSub(std::numeric_limits<std::int64_t>::max(), std::numeric_limits<std::int64_t>::min())
+         .has_value());
+static_assert(
+    !checkedSub(std::numeric_limits<std::int64_t>::min(), std::numeric_limits<std::int64_t>::max())
+         .has_value());
+
 /**
  * Calculate one number divided by another number in percentage.
  * The result is rounded up to the next integer, and capped in the range [0,100]

include/xrpl/config/Constants.h

@@ -11,6 +11,7 @@ struct Sections
     static constexpr auto kCompression = "compression";
     static constexpr auto kCrawl = "crawl";
     static constexpr auto kDatabasePath = "database_path";
+    static constexpr auto kDatagramMonitor = "datagram_monitor";
     static constexpr auto kDebugLogfile = "debug_logfile";
     static constexpr auto kElbSupport = "elb_support";
     static constexpr auto kFeatures = "features";

include/xrpl/protocol/detail/features.macro

@@ -15,6 +15,7 @@
 // Add new amendments to the top of this list.
 // Keep it sorted in reverse chronological order.
 
+XRPL_FIX    (BatchV1_2,                   Supported::Yes, VoteBehavior::DefaultYes)
 XRPL_FIX    (Cleanup3_4_0,                Supported::Yes, VoteBehavior::DefaultNo)
 XRPL_FEATURE(Sponsor,                     Supported::Yes, VoteBehavior::DefaultNo)
 XRPL_FEATURE(BatchV1_1,                   Supported::Yes, VoteBehavior::DefaultNo)

include/xrpl/server/NetworkOPs.h

@@ -12,6 +12,7 @@
 
 #include <boost/asio.hpp>
 
+#include <array>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
@@ -19,6 +20,7 @@
 #include <optional>
 #include <sstream>
 #include <string>
+#include <tuple>
 
 namespace xrpl {
 
@@ -83,6 +85,18 @@ class NetworkOPs : public InfoSub::Source
 public:
     using clock_type = beast::AbstractClock<std::chrono::steady_clock>;
 
+    // Snapshot of per-operating-mode accounting, exposed for the datagram monitor.
+    struct AccountingCounter
+    {
+        std::uint64_t transitions{0};
+        std::chrono::microseconds dur{std::chrono::microseconds(0)};
+    };
+    using StateAccountingData = std::tuple<
+        std::array<AccountingCounter, 5>,
+        OperatingMode,
+        std::chrono::steady_clock::time_point,
+        std::uint64_t>;
+
     enum class FailHard : unsigned char { No, Yes };
     static FailHard
     doFailHard(bool noMeansDont)
@@ -103,6 +117,8 @@ public:
 
     [[nodiscard]] virtual OperatingMode
     getOperatingMode() const = 0;
+    [[nodiscard]] virtual StateAccountingData
+    getStateAccountingData() = 0;
     [[nodiscard]] virtual std::string
     strOperatingMode(OperatingMode const mode, bool const admin = false) const = 0;
     [[nodiscard]] virtual std::string
@@ -220,6 +236,12 @@ public:
 
     virtual json::Value
     getConsensusInfo() = 0;
+    // Proposers and round time of the last consensus round, for out-of-band
+    // telemetry (DatagramMonitor) that cannot reach the private consensus object.
+    [[nodiscard]] virtual std::size_t
+    getPrevProposers() const = 0;
+    [[nodiscard]] virtual std::chrono::milliseconds
+    getPrevRoundTime() const = 0;
     virtual json::Value
     getServerInfo(bool human, bool admin, bool counters) = 0;
     virtual void

include/xrpl/tx/invariants/InvariantCheck.h

@@ -18,6 +18,8 @@
 #include <xrpl/tx/invariants/SponsorshipInvariant.h>
 #include <xrpl/tx/invariants/VaultInvariant.h>
 
+#include <boost/multiprecision/cpp_int.hpp>
+
 #include <cstdint>
 #include <set>
 #include <string>
@@ -139,7 +141,7 @@ public:
  */
 class XRPNotCreated
 {
-    std::int64_t drops_ = 0;
+    boost::multiprecision::int128_t drops_ = 0;
 
 public:
     void

include/xrpl/tx/paths/detail/Steps.h

@@ -1,6 +1,8 @@
 #pragma once
 
+#include <xrpl/basics/MathUtilities.h>
 #include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/contract.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/Zero.h>
 #include <xrpl/protocol/AccountID.h>
@@ -23,6 +25,7 @@
 #include <ostream>
 #include <stdexcept>
 #include <string>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -502,6 +505,46 @@ public:
 };
 /** @endcond */
 
+/** @cond INTERNAL */
+template <class T>
+[[nodiscard]] std::optional<T>
+checkedStepAddOpt(T const& lhs, T const& rhs)
+{
+    if constexpr (std::is_same_v<T, XRPAmount>)
+    {
+        if (auto const r = checkedAdd(lhs.drops(), rhs.drops()))
+            return XRPAmount{*r};
+        return std::nullopt;
+    }
+    else if constexpr (std::is_same_v<T, MPTAmount>)
+    {
+        if (auto const r = checkedAdd(lhs.value(), rhs.value()))
+            return MPTAmount{*r};
+        return std::nullopt;
+    }
+    else if constexpr (std::is_same_v<T, IOUAmount>)
+    {
+        // IOUAmount is Number-backed and throws on overflow.
+        return lhs + rhs;
+    }
+    else
+    {
+        // A new amount type must decide explicitly how to add; do not fall back
+        // to an unchecked add.
+        static_assert(sizeof(T) == 0, "checkedStepAddOpt: unsupported amount type");
+    }
+}
+
+template <class T>
+[[nodiscard]] T
+checkedStepAdd(T const& lhs, T const& rhs)
+{
+    if (auto const r = checkedStepAddOpt(lhs, rhs))
+        return *r;
+    Throw<FlowException>(tecPATH_DRY);
+}
+/** @endcond */
+
 /** @cond INTERNAL */
 // Check equal with tolerance
 bool

include/xrpl/tx/paths/detail/StrandFlow.h

@@ -31,7 +31,6 @@
 #include <cstdint>
 #include <iterator>
 #include <memory>
-#include <numeric>
 #include <optional>
 #include <tuple>
 #include <type_traits>
@@ -646,11 +645,21 @@ flow(
     boost::container::flat_multiset<TOutAmt> savedOuts;
     savedOuts.reserve(maxTries);
 
-    auto sum = [](auto const& col) {
+    // Returns std::nullopt if the aggregate overflows; callers treat that as a
+    // dry path.
+    auto sum = [](auto const& col) -> std::optional<std::decay_t<decltype(*col.begin())>> {
         using TResult = std::decay_t<decltype(*col.begin())>;
         if (col.empty())
             return TResult{beast::kZero};
-        return std::accumulate(col.begin() + 1, col.end(), *col.begin());
+        TResult total = *col.begin();
+        for (auto it = col.begin() + 1; it != col.end(); ++it)
+        {
+            auto const next = checkedStepAddOpt(total, *it);
+            if (!next)
+                return std::nullopt;
+            total = *next;
+        }
+        return total;
     };
 
     // These offers only need to be removed if the payment is not
@@ -749,9 +758,17 @@ flow(
         {
             savedIns.insert(best->in);
             savedOuts.insert(best->out);
-            remainingOut = outReq - sum(savedOuts);
+            auto const sumOut = sum(savedOuts);
+            if (!sumOut)
+                return {tecPATH_DRY, std::move(ofrsToRmOnFail)};
+            remainingOut = outReq - *sumOut;
             if (sendMax)
-                remainingIn = *sendMax - sum(savedIns);
+            {
+                auto const sumIn = sum(savedIns);
+                if (!sumIn)
+                    return {tecPATH_DRY, std::move(ofrsToRmOnFail)};
+                remainingIn = *sendMax - *sumIn;
+            }
 
             if (flowDebugInfo)
             {
@@ -786,8 +803,12 @@ flow(
             break;
     }
 
-    auto const actualOut = sum(savedOuts);
-    auto const actualIn = sum(savedIns);
+    auto const actualOutOpt = sum(savedOuts);
+    auto const actualInOpt = sum(savedIns);
+    if (!actualOutOpt || !actualInOpt)
+        return {tecPATH_DRY, std::move(ofrsToRmOnFail)};
+    auto const actualOut = *actualOutOpt;
+    auto const actualIn = *actualInOpt;
 
     JLOG(j.trace()) << "Total flow: in: " << to_string(actualIn)
                     << " out: " << to_string(actualOut);

src/libxrpl/ledger/helpers/TokenHelpers.cpp

@@ -1,6 +1,7 @@
 #include <xrpl/ledger/helpers/TokenHelpers.h>
 
 #include <xrpl/basics/Log.h>
+#include <xrpl/basics/MathUtilities.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/Zero.h>
 #include <xrpl/beast/utility/instrumentation.h>
@@ -1155,6 +1156,10 @@ accountSendMultiIOU(
 
         if (receiver)
         {
+            // Confirm the running debit will not overflow before crediting.
+            if (!checkedAdd(takeFromSender.xrp().drops(), amount.xrp().drops()))
+                return tecINTERNAL;
+
             // Increment XRP balance.
             auto const rcvBal = receiver->getFieldAmount(sfBalance);
             receiver->setFieldAmount(sfBalance, rcvBal + amount);
@@ -1162,7 +1167,7 @@ accountSendMultiIOU(
 
             view.update(receiver);
 
-            // Take what is actually sent
+            // Take what is actually sent.
             takeFromSender += amount;
         }
 
@@ -1438,6 +1443,8 @@ directSendNoLimitMultiMPT(
             }
 
             // Direct send: redeeming MPTs and/or sending own MPTs.
+            if (!checkedAdd(actual.mpt().value(), amount.mpt().value()))
+                return tecINTERNAL;
             if (auto const ter = directSendNoFeeMPT(view, senderID, receiverID, amount, j);
                 !isTesSuccess(ter))
                 return ter;
@@ -1451,6 +1458,10 @@ directSendNoLimitMultiMPT(
         STAmount const actualSend = (waiveFee == WaiveTransferFee::Yes)
             ? amount
             : multiply(amount, transferRate(view, amount.get<MPTIssue>().getMptID()));
+        // actual is a superset of takeFromSender, so checking it before both add
+        // sites also protects the debit accumulator.
+        if (!checkedAdd(actual.mpt().value(), actualSend.mpt().value()))
+            return tecINTERNAL;
         actual += actualSend;
         takeFromSender += actualSend;
 

src/libxrpl/protocol/BuildInfo.cpp

@@ -23,7 +23,7 @@ namespace {
 //------------------------------------------------------------------------------
 // clang-format off
 // NOLINTNEXTLINE(readability-identifier-naming)
-char const* const versionString = "3.4.0"
+char const* const versionString = "3.4.1"
     // clang-format on
     ;
 

src/libxrpl/tx/invariants/InvariantCheck.cpp

@@ -210,6 +210,9 @@ XRPNotCreated::finalize(
     ReadView const&,
     beast::Journal const& j) const
 {
+    // drops_ is a full-width running total, so a value that would have wrapped a
+    // 64-bit accumulator is caught here.
+
     // The net change should never be positive, as this would mean that the
     // transaction created XRP out of thin air. That's not possible.
     if (drops_ > 0)

src/libxrpl/tx/paths/BookStep.cpp

@@ -1060,7 +1060,10 @@ sum(TCollection const& col)
     using TResult = std::decay_t<decltype(*col.begin())>;
     if (col.empty())
         return TResult{beast::kZero};
-    return std::accumulate(col.begin() + 1, col.end(), *col.begin());
+    return std::accumulate(
+        col.begin() + 1, col.end(), *col.begin(), [](TResult const& a, TResult const& b) {
+            return checkedStepAdd(a, b);
+        });
 };
 
 template <class TIn, class TOut, class TDerived>

src/libxrpl/tx/transactors/system/Batch.cpp

@@ -9,6 +9,7 @@
 #include <xrpl/ledger/ReadView.h>
 #include <xrpl/ledger/helpers/SponsorHelpers.h>
 #include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Feature.h>
 #include <xrpl/protocol/Protocol.h>
 #include <xrpl/protocol/SField.h>
 #include <xrpl/protocol/STAccount.h>  // IWYU pragma: keep
@@ -194,6 +195,7 @@ Batch::getFlagsMask(PreflightContext const& ctx)
  * - The batch must contain at least two and no more than the maximum allowed
  * inner transactions.
  * - Each inner transaction must:
+ *   - Be wrapped in a RawTransaction object (with fixBatchV1_2).
  *   - Be unique within the batch.
  *   - Not itself be a Batch transaction.
  *   - Have the tfInnerBatchTxn flag set.
@@ -287,6 +289,14 @@ Batch::preflight(PreflightContext const& ctx)
     for (auto const& stxPtr : ctx.tx.getBatchTransactions())
     {
         STTx const& stx = *stxPtr;
+
+        if (ctx.rules.enabled(fixBatchV1_2) && stx.getFName() != sfRawTransaction)
+        {
+            JLOG(ctx.j.debug()) << "BatchTrace[" << parentBatchId << "]:"
+                                << "txns array may contain only RawTransaction objects.";
+            return temMALFORMED;
+        }
+
         auto const hash = stx.getTransactionID();
         if (!uniqueHashes.emplace(hash).second)
         {

src/test/app/Batch_test.cpp

@@ -5896,6 +5896,81 @@ class Batch_test : public beast::unit_test::Suite
         }
     }
 
+    void
+    testWrappedInnerSubmission(FeatureBitset features)
+    {
+        testcase("wrapper field submission");
+
+        using namespace test::jtx;
+
+        // Object fields with no InnerObjectFormats template, used in place of
+        // sfRawTransaction as the wrapper of each inner transaction.
+        for (SField const* wrapper :
+             {&sfRawTransaction,
+              &sfCreatedNode,
+              &sfModifiedNode,
+              &sfDeletedNode,
+              &sfTemplateEntry,
+              &sfEmitDetails,
+              &sfMemo,
+              &sfFinalFields,
+              &sfNewFields,
+              &sfPreviousFields,
+              &sfTransactionMetaData})
+        {
+            bool const poisoned = (wrapper != &sfRawTransaction);
+
+            auto const alice = Account("alice");
+            auto const bob = Account("bob");
+
+            auto wrap = [&](std::uint32_t s) {
+                json::Value inner = pay(alice, bob, XRP(1));
+                inner[jss::SigningPubKey] = "";
+                inner[jss::Sequence] = s;
+                inner[jss::Fee] = "0";
+                inner[jss::Flags] = tfInnerBatchTxn;
+
+                json::Value wrapped;
+                wrapped[wrapper->jsonName] = inner;
+                return wrapped;
+            };
+
+            auto submit = [&](Env& env, TER expected) {
+                env.fund(XRP(10000), alice, bob);
+                env.close();
+
+                auto const preBob = env.balance(bob);
+                auto const batchFee = batch::calcBatchFee(env, 0, 2);
+                auto const seq = env.seq(alice);
+
+                auto jv = batch::outer(alice, seq, batchFee, tfAllOrNothing);
+                jv[jss::RawTransactions][0u] = wrap(seq + 1);
+                jv[jss::RawTransactions][1u] = wrap(seq + 2);
+
+                env(jv, Ter(expected));
+                env.close();
+
+                return env.balance(bob) == preBob + XRP(2);
+            };
+
+            // Without fixBatchV1_2 every wrapper executes.
+            {
+                Env env{*this, features - fixBatchV1_2};
+                BEAST_EXPECTS(
+                    submit(env, tesSUCCESS),
+                    wrapper->getName() + " did not execute without fixBatchV1_2");
+            }
+
+            // With fixBatchV1_2 only sfRawTransaction is accepted.
+            {
+                Env env{*this, features | fixBatchV1_2};
+                bool const delivered = submit(env, poisoned ? TER{temMALFORMED} : TER{tesSUCCESS});
+                BEAST_EXPECTS(
+                    delivered == !poisoned, wrapper->getName() + " wrong result with fixBatchV1_2");
+            }
+        }
+    }
+
     void
     testWithFeats(FeatureBitset features)
     {
@@ -5935,6 +6010,7 @@ class Batch_test : public beast::unit_test::Suite
         testOuterBinding(features);
         testUnsortedBatchSigners(features);
         testBatchSigCache(features);
+        testWrappedInnerSubmission(features);
     }
 
 public:

src/test/core/Config_test.cpp

@@ -530,6 +530,25 @@ port_wss_admin
         }
     }
 
+    void
+    testDatagramMonitor()
+    {
+        testcase("datagram monitor");
+        Config withSink;
+        withSink.loadFromString(R"xrpldConfig(
+[datagram_monitor]
+127.0.0.1 9876
+)xrpldConfig");
+        BEAST_EXPECT(withSink.DATAGRAM_MONITOR == std::vector<std::string>{"127.0.0.1 9876"});
+
+        Config without;
+        without.loadFromString(R"xrpldConfig(
+[network_id]
+main
+)xrpldConfig");
+        BEAST_EXPECT(without.DATAGRAM_MONITOR.empty());
+    }
+
     void
     testNetworkID()
     {
@@ -1676,6 +1695,7 @@ r.ripple.com:51235
         testAmendment();
         testOverlay();
         testNetworkID();
+        testDatagramMonitor();
     }
 };
 

src/xrpld/app/main/Application.cpp

@@ -19,6 +19,7 @@
 #include <xrpld/app/main/LoadManager.h>
 #include <xrpld/app/main/NodeIdentity.h>
 #include <xrpld/app/main/NodeStoreScheduler.h>
+#include <xrpld/app/misc/DatagramMonitor.h>
 #include <xrpld/app/misc/SHAMapStore.h>
 #include <xrpld/app/misc/TxQ.h>
 #include <xrpld/app/misc/ValidatorKeys.h>
@@ -222,6 +223,7 @@ public:
     std::unique_ptr<JobQueue> jobQueue_;
     NodeStoreScheduler nodeStoreScheduler_;
     std::unique_ptr<SHAMapStore> shaMapStore_;
+    std::unique_ptr<DatagramMonitor> datagramMonitor_;
     PendingSaves pendingSaves_;
     std::optional<OpenLedger> openLedger_;
 
@@ -1525,6 +1527,14 @@ ApplicationImp::start(bool withTimers)
 
     ledgerCleaner_->start();
     perfLog_->start();
+
+    // Datagram monitor: UDP node-stats exporter (XDGM). Off in standalone or
+    // when [datagram_monitor] has no endpoints.
+    if (!config_->standalone() && !config_->DATAGRAM_MONITOR.empty())
+    {
+        datagramMonitor_ = std::make_unique<DatagramMonitor>(*this);
+        datagramMonitor_->start();
+    }
 }
 
 void
@@ -1543,6 +1553,12 @@ ApplicationImp::run()
 
     JLOG(journal_.debug()) << "Application stopping";
 
+    // Stop the datagram monitor before nodeStore_, networkOPs_, overlay_ and
+    // ledgerMaster_ stop below; datagramMonitor_ is declared ahead of them,
+    // so its implicit destruction would otherwise run after theirs.
+    if (datagramMonitor_)
+        datagramMonitor_->stop();
+
     io_latency_sampler_.cancelAsync();
 
     // VFALCO Enormous hack, we have to force the probe to cancel

src/xrpld/app/misc/DatagramMonitor.h

@@ -0,0 +1,906 @@
+#pragma once
+
+#include <xrpld/app/ledger/AcceptedLedger.h>
+#include <xrpld/app/ledger/InboundLedgers.h>
+#include <xrpld/app/ledger/LedgerMaster.h>
+#include <xrpld/app/main/Application.h>
+#include <xrpld/app/misc/ValidatorList.h>
+#include <xrpld/app/rdb/backend/SQLiteDatabase.h>
+#include <xrpld/overlay/Overlay.h>
+
+#include <xrpl/basics/UptimeClock.h>
+#include <xrpl/basics/mulDiv.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/ledger/CachedSLEs.h>
+#include <xrpl/nodestore/Database.h>
+#include <xrpl/protocol/BuildInfo.h>
+#include <xrpl/protocol/ErrorCodes.h>
+#include <xrpl/protocol/jss.h>
+#include <xrpl/server/LoadFeeTrack.h>
+#include <xrpl/server/NetworkOPs.h>
+
+#include <arpa/inet.h>
+#include <sys/resource.h>
+#include <sys/socket.h>
+
+#include <netdb.h>
+
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <cstring>
+#include <fstream>
+#include <sstream>
+#include <string>
+#if defined(__linux__)
+#include <sys/statvfs.h>
+#include <sys/sysinfo.h>
+#elif defined(__APPLE__)
+#include <mach/host_info.h>
+#include <mach/mach.h>
+#include <net/if.h>
+#include <net/if_dl.h>
+#include <sys/mount.h>
+#include <sys/sysctl.h>
+#include <sys/types.h>
+
+#include <ifaddrs.h>
+#endif
+#include <thread>
+#include <vector>
+
+namespace xrpl {
+
+// Magic number for server info packets: 'XDGM' (le) xrpld DataGram Monitor
+constexpr uint32_t SERVER_INFO_MAGIC = 0x4D474458;
+constexpr uint32_t SERVER_INFO_VERSION = 1;
+
+// Warning flag bits
+constexpr uint32_t WARNING_AMENDMENT_BLOCKED = 1 << 0;
+constexpr uint32_t WARNING_UNL_BLOCKED = 1 << 1;
+constexpr uint32_t WARNING_AMENDMENT_WARNED = 1 << 2;
+constexpr uint32_t WARNING_NOT_SYNCED = 1 << 3;
+
+// Time window statistics for rates
+struct [[gnu::packed]] MetricRates
+{
+    double rate_1m;   // Average rate over last minute
+    double rate_5m;   // Average rate over last 5 minutes
+    double rate_1h;   // Average rate over last hour
+    double rate_24h;  // Average rate over last 24 hours
+};
+
+struct AllRates
+{
+    MetricRates network_in;
+    MetricRates network_out;
+    MetricRates disk_read;
+    MetricRates disk_write;
+};
+
+// Structure to represent a ledger sequence range
+struct [[gnu::packed]] LgrRange
+{
+    uint32_t start;
+    uint32_t end;
+};
+
+// Map is returned separately since variable-length data
+// shouldn't be included in network structures
+using ObjectCountMap = std::vector<std::pair<std::basic_string<char>, int>>;
+
+struct [[gnu::packed]] DebugCounters
+{
+    // Database metrics
+    std::uint64_t dbKBTotal{0};
+    std::uint64_t dbKBLedger{0};
+    std::uint64_t dbKBTransaction{0};
+    std::uint64_t localTxCount{0};
+
+    // Basic metrics
+    std::uint32_t writeLoad{0};
+    std::int32_t historicalPerMinute{0};
+
+    // Cache metrics
+    std::uint32_t sleHitRate{0};     // Stored as fixed point, multiplied by 1000
+    std::uint32_t ledgerHitRate{0};  // Stored as fixed point, multiplied by 1000
+    std::uint32_t alSize{0};
+    std::uint32_t alHitRate{0};  // Stored as fixed point, multiplied by 1000
+    std::int32_t fullbelowSize{0};
+    std::uint32_t treenodeCacheSize{0};
+    std::uint32_t treenodeTrackSize{0};
+
+    // Node store metrics
+    std::uint64_t nodeWriteCount{0};
+    std::uint64_t nodeWriteSize{0};
+    std::uint64_t nodeFetchCount{0};
+    std::uint64_t nodeFetchHitCount{0};
+    std::uint64_t nodeFetchSize{0};
+
+    // Job queue metrics
+    std::uint64_t jobQueueOverflow{0};
+};
+
+// Core server metrics in the fixed header
+struct [[gnu::packed]] ServerInfoHeader
+{
+    // Fixed header fields come first
+    uint32_t magic;               // Magic number to identify packet type
+    uint32_t version;             // Protocol version number
+    uint32_t network_id;          // Network ID from config
+    uint32_t server_state;        // Operating mode as enum
+    uint32_t peer_count;          // Number of connected peers
+    uint32_t node_size;           // Size category (0=tiny through 4=huge)
+    uint32_t cpu_cores;           // CPU core count
+    uint32_t ledger_range_count;  // Number of range entries
+    uint32_t warning_flags;       // Warning flags (reduced size)
+
+    uint32_t padding_1;  // padding for alignment
+
+    // 64-bit metrics
+    uint64_t timestamp;          // System time in microseconds
+    uint64_t uptime;             // Server uptime in seconds
+    uint64_t io_latency_us;      // IO latency in microseconds
+    uint64_t validation_quorum;  // Validation quorum count
+    uint64_t fetch_pack_size;    // Size of fetch pack cache
+    uint64_t proposer_count;     // Number of proposers in last close
+    uint64_t converge_time_ms;   // Last convergence time in ms
+    uint64_t load_factor;        // Load factor (scaled by 1M)
+    uint64_t load_base;          // Load base value
+    uint64_t reserve_base;       // Reserve base amount
+    uint64_t reserve_inc;        // Reserve increment amount
+    uint64_t ledger_seq;         // Latest ledger sequence
+
+    // Fixed-size byte arrays
+    uint8_t ledger_hash[32];      // Latest ledger hash
+    uint8_t node_public_key[33];  // Node's public key
+    uint8_t padding2[7];          // Padding to maintain 8-byte alignment
+    uint8_t version_string[32];
+
+    // System metrics
+    uint64_t process_memory_pages;  // Process memory usage in bytes
+    uint64_t system_memory_total;   // Total system memory in bytes
+    uint64_t system_memory_free;    // Free system memory in bytes
+    uint64_t system_memory_used;    // Used system memory in bytes
+    uint64_t system_disk_total;     // Total disk space in bytes
+    uint64_t system_disk_free;      // Free disk space in bytes
+    uint64_t system_disk_used;      // Used disk space in bytes
+    uint64_t io_wait_time;          // IO wait time in milliseconds
+    double load_avg_1min;           // 1 minute load average
+    double load_avg_5min;           // 5 minute load average
+    double load_avg_15min;          // 15 minute load average
+
+    // State transition metrics
+    uint64_t state_transitions[5];  // Count for each operating mode
+    uint64_t state_durations[5];    // Duration in each mode
+    uint64_t initial_sync_us;       // Initial sync duration
+
+    // Network and disk rates remain unchanged
+    struct
+    {
+        MetricRates network_in;
+        MetricRates network_out;
+        MetricRates disk_read;
+        MetricRates disk_write;
+    } rates;
+
+    DebugCounters dbg_counters;
+};
+
+// System metrics collected for rate calculations
+struct SystemMetrics
+{
+    uint64_t timestamp;           // When metrics were collected
+    uint64_t network_bytes_in;    // Current total bytes in
+    uint64_t network_bytes_out;   // Current total bytes out
+    uint64_t disk_bytes_read;     // Current total bytes read
+    uint64_t disk_bytes_written;  // Current total bytes written
+};
+
+class MetricsTracker
+{
+private:
+    static constexpr size_t SAMPLES_1M = 60;     // 1 sample/second for 1 minute
+    static constexpr size_t SAMPLES_5M = 300;    // 1 sample/second for 5 minutes
+    static constexpr size_t SAMPLES_1H = 3600;   // 1 sample/second for 1 hour
+    static constexpr size_t SAMPLES_24H = 1440;  // 1 sample/minute for 24 hours
+
+    std::vector<SystemMetrics> samples_1m{SAMPLES_1M};
+    std::vector<SystemMetrics> samples_5m{SAMPLES_5M};
+    std::vector<SystemMetrics> samples_1h{SAMPLES_1H};
+    std::vector<SystemMetrics> samples_24h{SAMPLES_24H};
+
+    size_t index_1m{0}, index_5m{0}, index_1h{0}, index_24h{0};
+    std::chrono::system_clock::time_point last_24h_sample{};
+
+    double
+    calculateRate(
+        SystemMetrics const& current,
+        std::vector<SystemMetrics> const& samples,
+        size_t current_index,
+        size_t max_samples,
+        bool is_24h_window,
+        std::function<uint64_t(SystemMetrics const&)> metric_getter)
+    {
+        // If we don't have at least 2 samples, the rate is 0
+        if (current_index < 2)
+        {
+            return 0.0;
+        }
+
+        // Calculate time window based on the window type
+        uint64_t expected_window_micros;
+        if (is_24h_window)
+        {
+            expected_window_micros =
+                24ULL * 60ULL * 60ULL * 1000000ULL;  // 24 hours in microseconds
+        }
+        else
+        {
+            expected_window_micros =
+                max_samples * 1000000ULL;  // window in seconds * 1,000,000 for microseconds
+        }
+
+        // Get the oldest valid sample. Once the ring fills, the slot the
+        // next write will overwrite (current_index % max_samples) is the one
+        // that has gone longest without being refreshed.
+        size_t oldest_index = (current_index >= max_samples) ? (current_index % max_samples) : 0;
+        auto const& oldest = samples[oldest_index];
+
+        // For any window where we don't have full data, we should scale the
+        // rate based on the actual time we have data for
+        uint64_t actual_window_micros = current.timestamp - oldest.timestamp;
+        double window_scale =
+            std::min(1.0, static_cast<double>(actual_window_micros) / expected_window_micros);
+
+        double elapsed = actual_window_micros / 1000000.0;  // Convert microseconds to seconds
+
+        // Ensure we have a meaningful time difference
+        if (elapsed < 0.001)
+        {  // Less than 1ms difference
+            return 0.0;
+        }
+
+        uint64_t current_value = metric_getter(current);
+        uint64_t oldest_value = metric_getter(oldest);
+
+        // Handle counter wraparound
+        uint64_t diff = (current_value >= oldest_value)
+            ? (current_value - oldest_value)
+            : (std::numeric_limits<uint64_t>::max() - oldest_value + current_value + 1);
+
+        // Calculate the rate and scale it based on our window coverage
+        return (static_cast<double>(diff) / elapsed) * window_scale;
+    }
+
+    MetricRates
+    calculateMetricRates(
+        SystemMetrics const& current,
+        std::function<uint64_t(SystemMetrics const&)> metric_getter)
+    {
+        MetricRates rates;
+        rates.rate_1m =
+            calculateRate(current, samples_1m, index_1m, SAMPLES_1M, false, metric_getter);
+        rates.rate_5m =
+            calculateRate(current, samples_5m, index_5m, SAMPLES_5M, false, metric_getter);
+        rates.rate_1h =
+            calculateRate(current, samples_1h, index_1h, SAMPLES_1H, false, metric_getter);
+        rates.rate_24h =
+            calculateRate(current, samples_24h, index_24h, SAMPLES_24H, true, metric_getter);
+        return rates;
+    }
+
+public:
+    void
+    addSample(SystemMetrics const& metrics)
+    {
+        auto now = std::chrono::system_clock::now();
+
+        // Update 1-minute window (every second)
+        samples_1m[index_1m++ % SAMPLES_1M] = metrics;
+
+        // Update 5-minute window (every second)
+        samples_5m[index_5m++ % SAMPLES_5M] = metrics;
+
+        // Update 1-hour window (every second)
+        samples_1h[index_1h++ % SAMPLES_1H] = metrics;
+
+        // Update 24-hour window (every minute)
+        if (last_24h_sample + std::chrono::minutes(1) <= now)
+        {
+            samples_24h[index_24h++ % SAMPLES_24H] = metrics;
+            last_24h_sample = now;
+        }
+    }
+
+    AllRates
+    getRates(SystemMetrics const& current)
+    {
+        AllRates rates;
+        rates.network_in = calculateMetricRates(
+            current, [](SystemMetrics const& m) { return m.network_bytes_in; });
+        rates.network_out = calculateMetricRates(
+            current, [](SystemMetrics const& m) { return m.network_bytes_out; });
+        rates.disk_read =
+            calculateMetricRates(current, [](SystemMetrics const& m) { return m.disk_bytes_read; });
+        rates.disk_write = calculateMetricRates(
+            current, [](SystemMetrics const& m) { return m.disk_bytes_written; });
+        return rates;
+    }
+};
+
+class DatagramMonitor
+{
+private:
+    Application& app_;
+    beast::Journal j_;
+    std::atomic<bool> running_{false};
+    std::thread monitor_thread_;
+    MetricsTracker metrics_tracker_;
+
+    struct EndpointInfo
+    {
+        std::string ip;
+        uint16_t port;
+        bool is_ipv6;
+    };
+    EndpointInfo
+    parseEndpoint(std::string const& endpoint)
+    {
+        auto space_pos = endpoint.find(' ');
+        if (space_pos == std::string::npos)
+            throw std::runtime_error("Invalid endpoint format");
+
+        EndpointInfo info;
+        info.ip = endpoint.substr(0, space_pos);
+        info.port = std::stoi(endpoint.substr(space_pos + 1));
+        info.is_ipv6 = info.ip.find(':') != std::string::npos;
+        return info;
+    }
+
+    int
+    createSocket(EndpointInfo const& endpoint)
+    {
+        int sock = socket(endpoint.is_ipv6 ? AF_INET6 : AF_INET, SOCK_DGRAM, 0);
+        if (sock < 0)
+            throw std::runtime_error("Failed to create socket");
+        return sock;
+    }
+
+    void
+    sendPacket(int sock, EndpointInfo const& endpoint, std::vector<uint8_t> const& buffer)
+    {
+        struct sockaddr_storage addr;
+        socklen_t addr_len;
+
+        if (endpoint.is_ipv6)
+        {
+            struct sockaddr_in6* addr6 = reinterpret_cast<struct sockaddr_in6*>(&addr);
+            addr6->sin6_family = AF_INET6;
+            addr6->sin6_port = htons(endpoint.port);
+            inet_pton(AF_INET6, endpoint.ip.c_str(), &addr6->sin6_addr);
+            addr_len = sizeof(struct sockaddr_in6);
+        }
+        else
+        {
+            struct sockaddr_in* addr4 = reinterpret_cast<struct sockaddr_in*>(&addr);
+            addr4->sin_family = AF_INET;
+            addr4->sin_port = htons(endpoint.port);
+            inet_pton(AF_INET, endpoint.ip.c_str(), &addr4->sin_addr);
+            addr_len = sizeof(struct sockaddr_in);
+        }
+
+        sendto(
+            sock,
+            buffer.data(),
+            buffer.size(),
+            0,
+            reinterpret_cast<struct sockaddr*>(&addr),
+            addr_len);
+    }
+
+    // Returns both the counters and object count map separately
+    std::pair<DebugCounters, ObjectCountMap>
+    getDebugCounters()
+    {
+        DebugCounters counters;
+        ObjectCountMap objectCounts = CountedObjects::getInstance().getCounts(1);
+
+        // Database metrics if applicable
+        if (app_.config().useTxTables())
+        {
+            auto const db = dynamic_cast<SQLiteDatabase*>(&app_.getRelationalDatabase());
+            if (!db)
+                Throw<std::runtime_error>("Failed to get relational database");
+
+            if (auto dbKB = db->getKBUsedAll())
+                counters.dbKBTotal = dbKB;
+            if (auto dbKB = db->getKBUsedLedger())
+                counters.dbKBLedger = dbKB;
+            if (auto dbKB = db->getKBUsedTransaction())
+                counters.dbKBTransaction = dbKB;
+            if (auto count = app_.getOPs().getLocalTxCount())
+                counters.localTxCount = count;
+        }
+
+        // Basic metrics
+        counters.writeLoad = app_.getNodeStore().getWriteLoad();
+        counters.historicalPerMinute =
+            static_cast<std::int32_t>(app_.getInboundLedgers().fetchRate());
+
+        // Cache metrics - convert floating point rates to fixed point
+        counters.sleHitRate = 0;  // TODO: SLE cache hit-rate accessor absent on this fork
+        counters.ledgerHitRate =
+            static_cast<std::uint32_t>(app_.getLedgerMaster().getCacheHitRate() * 1000);
+        counters.alSize = app_.getAcceptedLedgerCache().size();
+        counters.alHitRate =
+            static_cast<std::uint32_t>(app_.getAcceptedLedgerCache().getHitRate() * 1000);
+        counters.fullbelowSize =
+            static_cast<std::int32_t>(app_.getNodeFamily().getFullBelowCache()->size());
+        counters.treenodeCacheSize = app_.getNodeFamily().getTreeNodeCache()->getCacheSize();
+        counters.treenodeTrackSize = app_.getNodeFamily().getTreeNodeCache()->getTrackSize();
+
+        // Get regular node store metrics
+        counters.nodeWriteCount = app_.getNodeStore().getStoreCount();
+        counters.nodeWriteSize = app_.getNodeStore().getStoreSize();
+        counters.nodeFetchCount = app_.getNodeStore().getFetchTotalCount();
+        counters.nodeFetchHitCount = app_.getNodeStore().getFetchHitCount();
+        counters.nodeFetchSize = app_.getNodeStore().getFetchSize();
+
+        counters.jobQueueOverflow = app_.getOverlay().getJqTransOverflow();
+
+        return {counters, objectCounts};
+    }
+
+    uint32_t
+    getPhysicalCPUCount()
+    {
+        static uint32_t count = 0;
+        if (count > 0)
+            return count;
+
+#if defined(__linux__)
+        try
+        {
+            std::ifstream cpuinfo("/proc/cpuinfo");
+            if (!cpuinfo)
+            {
+                JLOG(j_.error()) << "Unable to open file: /proc/cpuinfo";
+                return count;
+            }
+            std::string line;
+            std::set<std::string> physical_ids;
+            std::string current_physical_id;
+
+            while (std::getline(cpuinfo, line))
+            {
+                if (line.find("core id") != std::string::npos)
+                {
+                    current_physical_id = line.substr(line.find(":") + 1);
+                    // Trim whitespace
+                    current_physical_id.erase(0, current_physical_id.find_first_not_of(" \t"));
+                    current_physical_id.erase(current_physical_id.find_last_not_of(" \t") + 1);
+                    physical_ids.insert(current_physical_id);
+                }
+            }
+
+            count = physical_ids.size();
+        }
+        catch (std::exception const& e)
+        {
+            JLOG(j_.error()) << "Error getting CPU count: " << e.what();
+        }
+
+        // Return at least 1 if we couldn't determine the count
+        return count > 0 ? count : (count = 1);
+#elif defined(__APPLE__)
+        int value = 0;
+        size_t size = sizeof(value);
+        if (sysctlbyname("hw.physicalcpu", &value, &size, NULL, 0) == 0)
+            count = value;
+        return count > 0 ? count : (count = 1);
+#else
+        count = std::thread::hardware_concurrency();
+        return count > 0 ? count : (count = 1);
+#endif
+    }
+
+    SystemMetrics
+    collectSystemMetrics()
+    {
+        SystemMetrics metrics{};
+        metrics.timestamp = std::chrono::duration_cast<std::chrono::microseconds>(
+                                std::chrono::system_clock::now().time_since_epoch())
+                                .count();
+
+#if defined(__linux__)
+        // Network stats collection
+        try
+        {
+            std::ifstream net_file("/proc/net/dev");
+            if (!net_file)
+            {
+                JLOG(j_.error()) << "Unable to open file /proc/net/dev";
+                return metrics;
+            }
+
+            std::string line;
+            uint64_t total_bytes_in = 0, total_bytes_out = 0;
+
+            // Skip header lines
+            std::getline(net_file, line);  // Inter-|   Receive...
+            std::getline(net_file, line);  // face |bytes...
+
+            while (std::getline(net_file, line))
+            {
+                if (line.find(':') != std::string::npos)
+                {
+                    std::string interface = line.substr(0, line.find(':'));
+                    interface = interface.substr(interface.find_first_not_of(" \t"));
+                    interface = interface.substr(0, interface.find_last_not_of(" \t") + 1);
+
+                    // Skip loopback interface
+                    if (interface == "lo")
+                        continue;
+
+                    uint64_t bytes_in, bytes_out;
+                    std::istringstream iss(line.substr(line.find(':') + 1));
+                    iss >> bytes_in;  // rx bytes
+                    // Skip rx packets, errs, drop, fifo, frame, compressed, multicast
+                    uint64_t rx_field;
+                    for (int i = 0; i < 7; ++i)
+                        iss >> rx_field;
+                    iss >> bytes_out;  // tx bytes
+
+                    total_bytes_in += bytes_in;
+                    total_bytes_out += bytes_out;
+                }
+            }
+            metrics.network_bytes_in = total_bytes_in;
+            metrics.network_bytes_out = total_bytes_out;
+        }
+        catch (std::exception const& e)
+        {
+            JLOG(j_.error()) << "Error collecting network stats: " << e.what();
+        }
+
+        // Disk stats collection
+        try
+        {
+            std::ifstream disk_file("/proc/diskstats");
+            if (!disk_file)
+            {
+                JLOG(j_.error()) << "Unable to open file: /proc/diskstats";
+                return metrics;
+            }
+            std::string line;
+            uint64_t total_bytes_read = 0, total_bytes_written = 0;
+
+            while (std::getline(disk_file, line))
+            {
+                unsigned int major, minor;
+                char dev_name[32];
+                uint64_t reads, read_sectors, writes, write_sectors;
+
+                if (sscanf(
+                        line.c_str(),
+                        "%u %u %31s %lu %*u %lu %*u %lu %*u %lu",
+                        &major,
+                        &minor,
+                        dev_name,
+                        &reads,
+                        &read_sectors,
+                        &writes,
+                        &write_sectors) == 7)
+                {
+                    // Only process physical devices
+                    std::string device_name(dev_name);
+                    if (device_name.substr(0, 3) == "dm-" || device_name.substr(0, 4) == "loop" ||
+                        device_name.substr(0, 3) == "ram")
+                    {
+                        continue;
+                    }
+
+                    // Skip partitions (usually have a number at the end)
+                    if (std::isdigit(device_name.back()))
+                    {
+                        continue;
+                    }
+
+                    uint64_t bytes_read = read_sectors * 512;
+                    uint64_t bytes_written = write_sectors * 512;
+
+                    total_bytes_read += bytes_read;
+                    total_bytes_written += bytes_written;
+                }
+            }
+            metrics.disk_bytes_read = total_bytes_read;
+            metrics.disk_bytes_written = total_bytes_written;
+        }
+        catch (std::exception const& e)
+        {
+            JLOG(j_.error()) << "Error collecting disk stats: " << e.what();
+        }
+#elif defined(__APPLE__)
+        // Network stats collection
+        try
+        {
+            struct ifaddrs* ifap;
+            if (getifaddrs(&ifap) == 0)
+            {
+                uint64_t total_bytes_in = 0, total_bytes_out = 0;
+                for (struct ifaddrs* ifa = ifap; ifa; ifa = ifa->ifa_next)
+                {
+                    if (ifa->ifa_addr != NULL && ifa->ifa_addr->sa_family == AF_LINK)
+                    {
+                        struct if_data* ifd = (struct if_data*)ifa->ifa_data;
+                        if (ifd != NULL)
+                        {
+                            // Skip loopback interface
+                            if (strcmp(ifa->ifa_name, "lo0") == 0)
+                                continue;
+
+                            total_bytes_in += ifd->ifi_ibytes;
+                            total_bytes_out += ifd->ifi_obytes;
+                        }
+                    }
+                }
+                freeifaddrs(ifap);
+
+                metrics.network_bytes_in = total_bytes_in;
+                metrics.network_bytes_out = total_bytes_out;
+            }
+        }
+        catch (std::exception const& e)
+        {
+            JLOG(j_.error()) << "Error collecting network stats: " << e.what();
+        }
+
+        // Disk stats collection
+        // Disk IO stats are not easily accessible in macOS.
+        // We'll set these values to zero for now.
+        metrics.disk_bytes_read = 0;
+        metrics.disk_bytes_written = 0;
+#endif
+        return metrics;
+    }
+
+    std::vector<uint8_t>
+    generateServerInfo()
+    {
+        auto& ops = app_.getOPs();
+        auto& ledgerMaster = app_.getLedgerMaster();
+
+        auto currentMetrics = collectSystemMetrics();
+        metrics_tracker_.addSample(currentMetrics);
+
+        // Ledger ranges and the object-count map are omitted on this fork; the
+        // packet is the fixed header (core node + OS metrics + DebugCounters).
+        std::vector<uint8_t> buffer(sizeof(ServerInfoHeader));
+        auto* header = reinterpret_cast<ServerInfoHeader*>(buffer.data());
+        memset(header, 0, sizeof(ServerInfoHeader));
+
+        header->magic = SERVER_INFO_MAGIC;
+        header->version = SERVER_INFO_VERSION;
+        header->network_id = app_.config().networkId;
+        header->timestamp = std::chrono::duration_cast<std::chrono::microseconds>(
+                                std::chrono::system_clock::now().time_since_epoch())
+                                .count();
+        header->uptime = UptimeClock::now().time_since_epoch().count();
+        header->io_latency_us =
+            std::chrono::duration_cast<std::chrono::microseconds>(app_.getIOLatency()).count();
+        header->validation_quorum = app_.getValidators().quorum();
+        header->server_state = static_cast<std::uint32_t>(ops.getOperatingMode());
+        header->peer_count = app_.getOverlay().size();
+        header->node_size = app_.config().nodeSize;
+
+        auto const [counters, mode, start, initialSync] = ops.getStateAccountingData();
+        for (size_t i = 0; i < 5; ++i)
+        {
+            header->state_transitions[i] = counters[i].transitions;
+            header->state_durations[i] = counters[i].dur.count();
+        }
+        header->initial_sync_us = initialSync;
+
+        if (ops.isAmendmentBlocked())
+            header->warning_flags |= WARNING_AMENDMENT_BLOCKED;
+        if (ops.isUNLBlocked())
+            header->warning_flags |= WARNING_UNL_BLOCKED;
+        if (ops.isAmendmentWarned())
+            header->warning_flags |= WARNING_AMENDMENT_WARNED;
+        if (ops.getOperatingMode() != OperatingMode::FULL)
+            header->warning_flags |= WARNING_NOT_SYNCED;
+
+        header->proposer_count = ops.getPrevProposers();
+        header->converge_time_ms = ops.getPrevRoundTime().count();
+
+        auto const fp = ledgerMaster.getFetchPackCacheSize();
+        if (fp != 0)
+            header->fetch_pack_size = fp;
+
+        // Load factor (server only; fee-escalation term omitted on this fork).
+        header->load_factor = static_cast<std::uint64_t>(app_.getFeeTrack().getLoadFactor());
+        header->load_base = app_.getFeeTrack().getLoadBase();
+
+#if defined(__linux__)
+        // Get system info using sysinfo
+        struct sysinfo si;
+        if (sysinfo(&si) == 0)
+        {
+            header->system_memory_total = si.totalram * si.mem_unit;
+            header->system_memory_free = si.freeram * si.mem_unit;
+            header->system_memory_used = header->system_memory_total - header->system_memory_free;
+            header->load_avg_1min = si.loads[0] / (float)(1 << SI_LOAD_SHIFT);
+            header->load_avg_5min = si.loads[1] / (float)(1 << SI_LOAD_SHIFT);
+            header->load_avg_15min = si.loads[2] / (float)(1 << SI_LOAD_SHIFT);
+        }
+#elif defined(__APPLE__)
+        // Get total physical memory
+        int64_t physical_memory;
+        size_t length = sizeof(physical_memory);
+        if (sysctlbyname("hw.memsize", &physical_memory, &length, NULL, 0) == 0)
+        {
+            header->system_memory_total = physical_memory;
+        }
+
+        // Get free and used memory
+        vm_statistics_data_t vm_stats;
+        mach_msg_type_number_t count = HOST_VM_INFO_COUNT;
+        if (host_statistics(mach_host_self(), HOST_VM_INFO, (host_info_t)&vm_stats, &count) ==
+            KERN_SUCCESS)
+        {
+            uint64_t page_size;
+            length = sizeof(page_size);
+            sysctlbyname("hw.pagesize", &page_size, &length, NULL, 0);
+
+            header->system_memory_free = (uint64_t)vm_stats.free_count * page_size;
+            header->system_memory_used = header->system_memory_total - header->system_memory_free;
+        }
+
+        // Get load averages
+        double loadavg[3];
+        if (getloadavg(loadavg, 3) == 3)
+        {
+            header->load_avg_1min = loadavg[0];
+            header->load_avg_5min = loadavg[1];
+            header->load_avg_15min = loadavg[2];
+        }
+#endif
+
+        // Get process memory usage
+        struct rusage usage;
+        getrusage(RUSAGE_SELF, &usage);
+#if defined(__linux__)
+        // ru_maxrss is in KiB on Linux; macOS already reports bytes.
+        header->process_memory_pages = static_cast<uint64_t>(usage.ru_maxrss) * 1024;
+#else
+        header->process_memory_pages = static_cast<uint64_t>(usage.ru_maxrss);
+#endif
+
+        // Get disk usage
+#if defined(__linux__)
+        struct statvfs fs;
+        if (statvfs("/", &fs) == 0)
+        {
+            header->system_disk_total = fs.f_blocks * fs.f_frsize;
+            header->system_disk_free = fs.f_bfree * fs.f_frsize;
+            header->system_disk_used = header->system_disk_total - header->system_disk_free;
+        }
+#elif defined(__APPLE__)
+        struct statfs fs;
+        if (statfs("/", &fs) == 0)
+        {
+            header->system_disk_total = fs.f_blocks * fs.f_bsize;
+            header->system_disk_free = fs.f_bfree * fs.f_bsize;
+            header->system_disk_used = header->system_disk_total - header->system_disk_free;
+        }
+#endif
+
+        // Get CPU core count
+        header->cpu_cores = getPhysicalCPUCount();
+
+        // Get rate statistics
+        auto rates = metrics_tracker_.getRates(currentMetrics);
+        header->rates.network_in = rates.network_in;
+        header->rates.network_out = rates.network_out;
+        header->rates.disk_read = rates.disk_read;
+        header->rates.disk_write = rates.disk_write;
+
+        // Ledger height + hash via stable accessors (this fork's Ledger lacks
+        // info()). The hash lets the collector detect a fork: divergent
+        // ledger_hash across nodes at the same ledger_seq.
+        std::uint32_t const validSeq = ledgerMaster.getValidLedgerIndex();
+        header->ledger_seq = validSeq;
+        uint256 const validHash = ledgerMaster.getHashBySeq(validSeq);
+        std::memcpy(header->ledger_hash, validHash.data(), 32);
+        header->reserve_base = app_.config().fees.accountReserve.drops();
+        header->reserve_inc = app_.config().fees.ownerReserve.drops();
+
+        // Node public key + version string.
+        auto const& nodeKey = app_.nodeIdentity().first;
+        std::memcpy(header->node_public_key, nodeKey.data(), 33);
+        memset(&header->version_string, 0, 32);
+        memcpy(
+            &header->version_string,
+            build_info::getVersionString().c_str(),
+            build_info::getVersionString().size() > 32 ? 32
+                                                       : build_info::getVersionString().size());
+
+        header->ledger_range_count = 0;
+        header->dbg_counters = getDebugCounters().first;
+        return buffer;
+    }
+    void
+    monitorThread()
+    {
+        std::vector<std::pair<EndpointInfo, int>> endpoints;
+
+        try
+        {
+            for (auto const& epStr : app_.config().DATAGRAM_MONITOR)
+            {
+                auto endpoint = parseEndpoint(epStr);
+                endpoints.push_back(std::make_pair(endpoint, createSocket(endpoint)));
+            }
+        }
+        catch (std::exception const& e)
+        {
+            JLOG(j_.error()) << "Invalid [datagram_monitor] configuration: " << e.what();
+            running_ = false;
+            return;
+        }
+
+        while (running_)
+        {
+            try
+            {
+                auto info = generateServerInfo();
+                for (auto const& ep : endpoints)
+                {
+                    sendPacket(ep.second, ep.first, info);
+                }
+                std::this_thread::sleep_for(std::chrono::seconds(1));
+            }
+            catch (std::exception const& e)
+            {
+                // Log error but continue monitoring
+                JLOG(j_.error()) << "Server info monitor error: " << e.what();
+            }
+        }
+
+        for (auto const& ep : endpoints)
+        {
+            close(ep.second);
+        }
+    }
+
+public:
+    DatagramMonitor(Application& app) : app_(app), j_(beast::Journal::getNullSink())
+    {
+    }
+
+    void
+    start()
+    {
+        if (!running_.exchange(true))
+        {
+            monitor_thread_ = std::thread(&DatagramMonitor::monitorThread, this);
+        }
+    }
+
+    void
+    stop()
+    {
+        if (running_.exchange(false))
+        {
+            if (monitor_thread_.joinable())
+                monitor_thread_.join();
+        }
+    }
+
+    ~DatagramMonitor()
+    {
+        stop();
+    }
+};
+}  // namespace xrpl

src/xrpld/app/misc/NetworkOPs.cpp

@@ -356,6 +356,9 @@ public:
     OperatingMode
     getOperatingMode() const override;
 
+    StateAccountingData
+    getStateAccountingData() override;
+
     std::string
     strOperatingMode(OperatingMode const mode, bool const admin) const override;
 
@@ -521,6 +524,10 @@ public:
 
     json::Value
     getConsensusInfo() override;
+    std::size_t
+    getPrevProposers() const override;
+    std::chrono::milliseconds
+    getPrevRoundTime() const override;
     json::Value
     getServerInfo(bool human, bool admin, bool counters) override;
     void
@@ -1091,6 +1098,16 @@ NetworkOPsImp::getOperatingMode() const
     return mode_;
 }
 
+NetworkOPs::StateAccountingData
+NetworkOPsImp::getStateAccountingData()
+{
+    auto const data = accounting_.getCounterData();
+    std::array<NetworkOPs::AccountingCounter, 5> out;
+    for (std::size_t i = 0; i < out.size(); ++i)
+        out[i] = {data.counters[i].transitions, data.counters[i].dur};
+    return {out, data.mode, data.start, data.initialSyncUs};
+}
+
 inline std::string
 NetworkOPsImp::strOperatingMode(bool const admin /* = false */) const
 {
@@ -2814,6 +2831,18 @@ NetworkOPsImp::getConsensusInfo()
     return consensus_.getJson(true);
 }
 
+std::size_t
+NetworkOPsImp::getPrevProposers() const
+{
+    return consensus_.prevProposers();
+}
+
+std::chrono::milliseconds
+NetworkOPsImp::getPrevRoundTime() const
+{
+    return consensus_.prevRoundTime();
+}
+
 json::Value
 NetworkOPsImp::getServerInfo(bool human, bool admin, bool counters)
 {

src/xrpld/core/Config.h

@@ -150,6 +150,10 @@ public:
     // Entries from [ips_fixed] config stanza
     std::vector<std::string> ipsFixed;
 
+    // Entries from [datagram_monitor]: "<IP> <port>" UDP targets the
+    // DatagramMonitor sends node-stats packets to (XDGM, every 1s).
+    std::vector<std::string> DATAGRAM_MONITOR;
+
     StartUpType startUp = StartUpType::Normal;
 
     bool startValid = false;

src/xrpld/core/detail/Config.cpp

@@ -481,6 +481,9 @@ Config::loadFromString(std::string const& fileContents)
     if (auto s = getIniFileSection(secConfig, Sections::kIpsFixed))
         ipsFixed = *s;
 
+    if (auto s = getIniFileSection(secConfig, Sections::kDatagramMonitor))
+        DATAGRAM_MONITOR = *s;
+
     // if the user has specified ip:port then replace : with a space.
     {
         auto replaceColons = [](std::vector<std::string>& strVec) {