Do not allow mempool clusters to exceed configured limits

diff --git a/src/test/mempool_tests.cpp b/src/test/mempool_tests.cpp
index 5f8b63db..baee31e8 100644
--- a/src/test/mempool_tests.cpp
+++ b/src/test/mempool_tests.cpp
@@ -547,7 +547,7 @@ BOOST_AUTO_TEST_CASE(MempoolSizeLimitTest)
         AddToMempool(pool, entry.Fee(100LL).FromTx(tx5));
     AddToMempool(pool, entry.Fee(900LL).FromTx(tx7));
 
-    pool.TrimToSize(pool.DynamicMemoryUsage() / 2); // should maximize mempool size by only removing 5/7
+    pool.TrimToSize(pool.DynamicMemoryUsage() * 0.75); // should maximize mempool size by only removing 5/7
     BOOST_CHECK(pool.exists(tx4.GetHash()));
     BOOST_CHECK(!pool.exists(tx5.GetHash()));
     BOOST_CHECK(pool.exists(tx6.GetHash()));
diff --git a/src/txmempool.cpp b/src/txmempool.cpp
index dc8ec8ed..af3a54cc 100644
--- a/src/txmempool.cpp
+++ b/src/txmempool.cpp
@@ -145,11 +145,20 @@ void CTxMemPool::UpdateTransactionsFromBlock(const std::vector<Txid>& vHashesToU
                     UpdateChild(it, childIter, true);
                     UpdateParent(childIter, it, true);
                 }
+                // Add dependencies that are discovered between transactions in the
+                // block and transactions that were in the mempool to txgraph.
+                m_txgraph->AddDependency(/*parent=*/*it, /*child=*/*childIter);
             }
         } // release epoch guard for UpdateForDescendants
         UpdateForDescendants(it, mapMemPoolDescendantsToUpdate, setAlreadyIncluded, descendants_to_remove);
     }
 
+    auto txs_to_remove = m_txgraph->Trim(); // Enforce cluster size limits.
+    for (auto txptr : txs_to_remove) {
+        const CTxMemPoolEntry& entry = *(static_cast<const CTxMemPoolEntry*>(txptr));
+        descendants_to_remove.insert(entry.GetTx().GetHash());
+    }
+
     for (const auto& txid : descendants_to_remove) {
         // This txid may have been removed already in a prior call to removeRecursive.
         // Therefore we ensure it is not yet removed already.
@@ -229,6 +238,7 @@ util::Result<void> CTxMemPool::CheckPackageLimits(const Package& package,
             }
         }
     }
+
     // When multiple transactions are passed in, the ancestors and descendants of all transactions
     // considered together must be within limits even if they are not interdependent. This may be
     // stricter than the limits for each individual transaction.
@@ -413,7 +423,7 @@ static CTxMemPool::Options&& Flatten(CTxMemPool::Options&& opts, bilingual_str&
 CTxMemPool::CTxMemPool(Options opts, bilingual_str& error)
     : m_opts{Flatten(std::move(opts), error)}
 {
-    m_txgraph = MakeTxGraph(64, 101'000, ACCEPTABLE_ITERS);
+    m_txgraph = MakeTxGraph(m_opts.limits.cluster_count, m_opts.limits.cluster_size_vbytes * WITNESS_SCALE_FACTOR, ACCEPTABLE_ITERS);
 }
 
 bool CTxMemPool::isSpent(const COutPoint& outpoint) const
@@ -1392,6 +1402,10 @@ CTxMemPool::ChangeSet::TxHandle CTxMemPool::ChangeSet::StageAddition(const CTran
 {
     LOCK(m_pool->cs);
     Assume(m_to_add.find(tx->GetHash()) == m_to_add.end());
+    Assume(!m_dependencies_processed);
+
+    // We need to process dependencies after adding a new transaction.
+    m_dependencies_processed = false;
 
     CAmount delta{0};
     m_pool->ApplyDelta(tx->GetHash(), delta);
@@ -1417,9 +1431,44 @@ void CTxMemPool::ChangeSet::StageRemoval(CTxMemPool::txiter it)
 void CTxMemPool::ChangeSet::Apply()
 {
     LOCK(m_pool->cs);
+    if (!m_dependencies_processed) {
+        ProcessDependencies();
+    }
     m_pool->Apply(this);
     m_to_add.clear();
     m_to_remove.clear();
     m_entry_vec.clear();
     m_ancestors.clear();
 }
+
+void CTxMemPool::ChangeSet::ProcessDependencies()
+{
+    LOCK(m_pool->cs);
+    Assume(!m_dependencies_processed); // should only call this once.
+    for (const auto& entryptr : m_entry_vec) {
+        for (const auto &txin : entryptr->GetSharedTx()->vin) {
+            std::optional<txiter> piter = m_pool->GetIter(txin.prevout.hash);
+            if (!piter) {
+                auto it = m_to_add.find(txin.prevout.hash);
+                if (it != m_to_add.end()) {
+                    piter = std::make_optional(it);
+                }
+            }
+            if (piter) {
+                m_pool->m_txgraph->AddDependency(/*parent=*/**piter, /*child=*/*entryptr);
+            }
+        }
+    }
+    m_dependencies_processed = true;
+    return;
+ }
+
+bool CTxMemPool::ChangeSet::CheckMemPoolPolicyLimits()
+{
+    LOCK(m_pool->cs);
+    if (!m_dependencies_processed) {
+        ProcessDependencies();
+    }
+
+    return !m_pool->m_txgraph->IsOversized(TxGraph::Level::TOP);
+}
diff --git a/src/txmempool.h b/src/txmempool.h
index acf55984..60087561 100644
--- a/src/txmempool.h
+++ b/src/txmempool.h
@@ -842,6 +842,9 @@ public:
 
         const CTxMemPool::setEntries& GetRemovals() const { return m_to_remove; }
 
+        /** Check if any cluster limits are exceeded. Returns true if pass, false if fail. */
+        bool CheckMemPoolPolicyLimits();
+
         util::Result<CTxMemPool::setEntries> CalculateMemPoolAncestors(TxHandle tx, const Limits& limits)
         {
             // Look up transaction in our cache first
@@ -879,12 +882,15 @@ public:
         void Apply() EXCLUSIVE_LOCKS_REQUIRED(cs_main);
 
     private:
+        void ProcessDependencies();
+
         CTxMemPool* m_pool;
         CTxMemPool::indexed_transaction_set m_to_add;
         std::vector<CTxMemPool::txiter> m_entry_vec; // track the added transactions' insertion order
         // map from the m_to_add index to the ancestors for the transaction
         std::map<CTxMemPool::txiter, CTxMemPool::setEntries, CompareIteratorByHash> m_ancestors;
         CTxMemPool::setEntries m_to_remove;
+        bool m_dependencies_processed{false};
 
         friend class CTxMemPool;
     };
diff --git a/src/validation.cpp b/src/validation.cpp
index 25831db6..596f688b 100644
--- a/src/validation.cpp
+++ b/src/validation.cpp
@@ -1126,6 +1126,7 @@ bool MemPoolAccept::ReplacementChecks(Workspace& ws)
         return state.Invalid(TxValidationResult::TX_MEMPOOL_POLICY,
                              strprintf("too many potential replacements%s", ws.m_sibling_eviction ? " (including sibling eviction)" : ""), *err_string);
     }
+
     // Enforce Rule #2.
     if (const auto err_string{HasNoNewUnconfirmed(tx, m_pool, all_conflicts)}) {
         // Sibling eviction is only done for TRUC transactions, which cannot have multiple ancestors.
@@ -1460,13 +1461,6 @@ MempoolAcceptResult MemPoolAccept::AcceptSingleTransactionInternal(const CTransa
         return MempoolAcceptResult::Failure(ws.m_state);
     }
 
-    if (m_pool.m_opts.require_standard) {
-        Wtxid dummy_wtxid;
-        if (!CheckEphemeralSpends(/*package=*/{ptx}, m_pool.m_opts.dust_relay_feerate, m_pool, ws.m_state, dummy_wtxid)) {
-            return MempoolAcceptResult::Failure(ws.m_state);
-        }
-    }
-
     if (m_subpackage.m_rbf && !ReplacementChecks(ws)) {
         if (ws.m_state.GetResult() == TxValidationResult::TX_RECONSIDERABLE) {
             // Failed for incentives-based fee reasons. Provide the effective feerate and which tx was included.
@@ -1475,6 +1469,19 @@ MempoolAcceptResult MemPoolAccept::AcceptSingleTransactionInternal(const CTransa
         return MempoolAcceptResult::Failure(ws.m_state);
     }
 
+    // Check if the transaction would exceed the cluster size limit.
+    if (!m_subpackage.m_changeset->CheckMemPoolPolicyLimits()) {
+        ws.m_state.Invalid(TxValidationResult::TX_MEMPOOL_POLICY, "too-large-cluster", "");
+        return MempoolAcceptResult::Failure(ws.m_state);
+    }
+
+    if (m_pool.m_opts.require_standard) {
+        Wtxid dummy_wtxid;
+        if (!CheckEphemeralSpends(/*package=*/{ptx}, m_pool.m_opts.dust_relay_feerate, m_pool, ws.m_state, dummy_wtxid)) {
+            return MempoolAcceptResult::Failure(ws.m_state);
+        }
+    }
+
     // Perform the inexpensive checks first and avoid hashing and signature verification unless
     // those checks pass, to mitigate CPU exhaustion denial-of-service attacks.
     if (!PolicyScriptChecks(args, ws)) return MempoolAcceptResult::Failure(ws.m_state);
@@ -1615,6 +1622,12 @@ PackageMempoolAcceptResult MemPoolAccept::AcceptMultipleTransactionsInternal(con
         return PackageMempoolAcceptResult(package_state, std::move(results));
     }
 
+    // Check if the transactions would exceed the cluster size limit.
+    if (!m_subpackage.m_changeset->CheckMemPoolPolicyLimits()) {
+        package_state.Invalid(PackageValidationResult::PCKG_POLICY, "too-large-cluster", "");
+        return PackageMempoolAcceptResult(package_state, std::move(results));
+    }
+
     // Now that we've bounded the resulting possible ancestry count, check package for dust spends
     if (m_pool.m_opts.require_standard) {
         TxValidationState child_state;
diff --git a/test/functional/mempool_ephemeral_dust.py b/test/functional/mempool_ephemeral_dust.py
index a0308da7..712176e9 100755
--- a/test/functional/mempool_ephemeral_dust.py
+++ b/test/functional/mempool_ephemeral_dust.py
@@ -419,7 +419,8 @@ class EphemeralDustTest(BitcoinTestFramework):
 
         res = self.nodes[0].submitpackage([dusty_tx["hex"] for dusty_tx in dusty_txs] + [insufficient_sweep_tx["hex"]])
         assert_equal(res['package_msg'], "transaction failed")
-        assert_equal(res['tx-results'][insufficient_sweep_tx["wtxid"]]["error"], f"missing-ephemeral-spends, tx {insufficient_sweep_tx['txid']} (wtxid={insufficient_sweep_tx['wtxid']}) did not spend parent's ephemeral dust")
+        #assert_equal(res['tx-results'][insufficient_sweep_tx["wtxid"]]["error"], f"missing-ephemeral-spends, tx {insufficient_sweep_tx['txid']} (wtxid={insufficient_sweep_tx['wtxid']}) did not spend parent's ephemeral dust")
+        assert_equal(res['tx-results'][insufficient_sweep_tx["wtxid"]]["error"], f"replacement-adds-unconfirmed, replacement {insufficient_sweep_tx['txid']} adds unconfirmed input, idx 46")
         assert_mempool_contents(self, self.nodes[0], expected=[dusty_tx["tx"] for dusty_tx in dusty_txs] + [sweep_all_but_one_tx["tx"]])
 
         # Cycle out the partial sweep to avoid triggering package RBF behavior which limits package to no in-mempool ancestors
diff --git a/test/functional/mempool_updatefromblock.py b/test/functional/mempool_updatefromblock.py
index fdc6b9a3..6ee31ffa 100755
--- a/test/functional/mempool_updatefromblock.py
+++ b/test/functional/mempool_updatefromblock.py
@@ -28,7 +28,7 @@ class MempoolUpdateFromBlockTest(BitcoinTestFramework):
     def set_test_params(self):
         self.num_nodes = 1
         # Ancestor and descendant limits depend on transaction_graph_test requirements
-        self.extra_args = [['-limitdescendantsize=1000', '-limitancestorsize=1000', f'-limitancestorcount={CUSTOM_ANCESTOR_COUNT}', f'-limitdescendantcount={CUSTOM_DESCENDANT_COUNT}']]
+        self.extra_args = [['-limitdescendantsize=1000', '-limitancestorsize=1000', f'-limitancestorcount={CUSTOM_ANCESTOR_COUNT}', f'-limitdescendantcount={CUSTOM_DESCENDANT_COUNT}', "-limitclustersize=1000"]]
 
     def create_empty_fork(self, fork_length):
         '''
@@ -198,11 +198,11 @@ class MempoolUpdateFromBlockTest(BitcoinTestFramework):
         fork_blocks = self.create_empty_fork(fork_length=10)
 
         # Two higher than descendant count
-        chain = wallet.create_self_transfer_chain(chain_length=CUSTOM_DESCENDANT_COUNT + 2)
+        chain = wallet.create_self_transfer_chain(chain_length=64 + 2)
         for tx in chain[:-2]:
             self.nodes[0].sendrawtransaction(tx["hex"])
 
-        assert_raises_rpc_error(-26, "too-long-mempool-chain, too many unconfirmed ancestors [limit: 100]", self.nodes[0].sendrawtransaction, chain[-2]["hex"])
+        assert_raises_rpc_error(-26, "too-large-cluster", self.nodes[0].sendrawtransaction, chain[-2]["hex"])
 
         # Mine a block with all but last transaction, non-standardly long chain
         self.generateblock(self.nodes[0], output="raw(42)", transactions=[tx["hex"] for tx in chain[:-1]])
@@ -220,7 +220,7 @@ class MempoolUpdateFromBlockTest(BitcoinTestFramework):
 
     def run_test(self):
         # Mine in batches of 25 to test multi-block reorg under chain limits
-        self.transaction_graph_test(size=CUSTOM_ANCESTOR_COUNT, n_tx_to_mine=[25, 50, 75])
+        self.transaction_graph_test(size=64, n_tx_to_mine=[25, 50, 75])
 
         self.test_max_disconnect_pool_bytes()